cubefs

1
package sarama
2

3
type EndTxnRequest struct {
4
	TransactionalID   string
5
	ProducerID        int64
6
	ProducerEpoch     int16
7
	TransactionResult bool
8
}
9

10
func (a *EndTxnRequest) encode(pe packetEncoder) error {
11
	if err := pe.putString(a.TransactionalID); err != nil {
12
		return err
13
	}
14

15
	pe.putInt64(a.ProducerID)
16

17
	pe.putInt16(a.ProducerEpoch)
18

19
	pe.putBool(a.TransactionResult)
20

21
	return nil
22
}
23

24
func (a *EndTxnRequest) decode(pd packetDecoder, version int16) (err error) {
25
	if a.TransactionalID, err = pd.getString(); err != nil {
26
		return err
27
	}
28
	if a.ProducerID, err = pd.getInt64(); err != nil {
29
		return err
30
	}
31
	if a.ProducerEpoch, err = pd.getInt16(); err != nil {
32
		return err
33
	}
34
	if a.TransactionResult, err = pd.getBool(); err != nil {
35
		return err
36
	}
37
	return nil
38
}
39

40
func (a *EndTxnRequest) key() int16 {
41
	return 26
42
}
43

44
func (a *EndTxnRequest) version() int16 {
45
	return 0
46
}
47

48
func (r *EndTxnRequest) headerVersion() int16 {
49
	return 1
50
}
51

52
func (a *EndTxnRequest) requiredVersion() KafkaVersion {
53
	return V0_11_0_0
54
}
55