add_partitions_to_txn_request.go 1.5 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677
  1. package sarama
  2. //AddPartitionsToTxnRequest is a add paartition request
  3. type AddPartitionsToTxnRequest struct {
  4. TransactionalID string
  5. ProducerID int64
  6. ProducerEpoch int16
  7. TopicPartitions map[string][]int32
  8. }
  9. func (a *AddPartitionsToTxnRequest) encode(pe packetEncoder) error {
  10. if err := pe.putString(a.TransactionalID); err != nil {
  11. return err
  12. }
  13. pe.putInt64(a.ProducerID)
  14. pe.putInt16(a.ProducerEpoch)
  15. if err := pe.putArrayLength(len(a.TopicPartitions)); err != nil {
  16. return err
  17. }
  18. for topic, partitions := range a.TopicPartitions {
  19. if err := pe.putString(topic); err != nil {
  20. return err
  21. }
  22. if err := pe.putInt32Array(partitions); err != nil {
  23. return err
  24. }
  25. }
  26. return nil
  27. }
  28. func (a *AddPartitionsToTxnRequest) decode(pd packetDecoder, version int16) (err error) {
  29. if a.TransactionalID, err = pd.getString(); err != nil {
  30. return err
  31. }
  32. if a.ProducerID, err = pd.getInt64(); err != nil {
  33. return err
  34. }
  35. if a.ProducerEpoch, err = pd.getInt16(); err != nil {
  36. return err
  37. }
  38. n, err := pd.getArrayLength()
  39. if err != nil {
  40. return err
  41. }
  42. a.TopicPartitions = make(map[string][]int32)
  43. for i := 0; i < n; i++ {
  44. topic, err := pd.getString()
  45. if err != nil {
  46. return err
  47. }
  48. partitions, err := pd.getInt32Array()
  49. if err != nil {
  50. return err
  51. }
  52. a.TopicPartitions[topic] = partitions
  53. }
  54. return nil
  55. }
  56. func (a *AddPartitionsToTxnRequest) key() int16 {
  57. return 24
  58. }
  59. func (a *AddPartitionsToTxnRequest) version() int16 {
  60. return 0
  61. }
  62. func (a *AddPartitionsToTxnRequest) requiredVersion() KafkaVersion {
  63. return V0_11_0_0
  64. }