Skip to content

Commit af1725f

Browse files
authored
Fix: Update OffsetCommit CommittedLeaderEpoch to v6 (#1335)
* Update OffsetCommit CommittedLeaderEpoch to v6, fixes #1334 * update OffsetCommitRequest test to use correct protocol version
1 parent 3683dc8 commit af1725f

File tree

2 files changed

+36
-12
lines changed

2 files changed

+36
-12
lines changed

protocol/offsetcommit/offsetcommit.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ type RequestPartition struct {
2828
PartitionIndex int32 `kafka:"min=v0,max=v7"`
2929
CommittedOffset int64 `kafka:"min=v0,max=v7"`
3030
CommitTimestamp int64 `kafka:"min=v1,max=v1"`
31-
CommittedLeaderEpoch int32 `kafka:"min=v5,max=v7"`
31+
CommittedLeaderEpoch int32 `kafka:"min=v6,max=v7"`
3232
CommittedMetadata string `kafka:"min=v0,max=v7,nullable"`
3333
}
3434

protocol/offsetcommit/offsetcommit_test.go

Lines changed: 35 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -93,30 +93,54 @@ func TestOffsetCommitRequest(t *testing.T) {
9393
})
9494
}
9595

96-
// Version 5 added:
97-
// RequestTopic.RequestPartition.CommitedLeaderEpoc
9896
// Version 5 removed:
9997
// RetentionTimeMs
100-
// Fields are the same through version 6.
101-
for _, version := range []int16{5, 6} {
98+
for _, version := range []int16{5} {
10299
prototest.TestRequest(t, version, &offsetcommit.Request{
103100
GroupID: "group-3",
104101
GenerationID: 1,
105102
MemberID: "member-3",
106103
Topics: []offsetcommit.RequestTopic{
107104
{
108105
Name: "topic-3",
106+
Partitions: []offsetcommit.RequestPartition{
107+
{
108+
PartitionIndex: 0,
109+
CommittedOffset: 1,
110+
CommittedMetadata: "meta-3-0",
111+
},
112+
{
113+
PartitionIndex: 1,
114+
CommittedOffset: 2,
115+
CommittedMetadata: "meta-3-1",
116+
},
117+
},
118+
},
119+
},
120+
})
121+
}
122+
123+
// Version 6 added:
124+
// RequestTopic.RequestPartition.CommittedLeaderEpoch
125+
for _, version := range []int16{6} {
126+
prototest.TestRequest(t, version, &offsetcommit.Request{
127+
GroupID: "group-4",
128+
GenerationID: 1,
129+
MemberID: "member-4",
130+
Topics: []offsetcommit.RequestTopic{
131+
{
132+
Name: "topic-4",
109133
Partitions: []offsetcommit.RequestPartition{
110134
{
111135
PartitionIndex: 0,
112136
CommittedOffset: 1,
113-
CommittedMetadata: "meta-3-0",
137+
CommittedMetadata: "meta-4-0",
114138
CommittedLeaderEpoch: 10,
115139
},
116140
{
117141
PartitionIndex: 1,
118142
CommittedOffset: 2,
119-
CommittedMetadata: "meta-3-1",
143+
CommittedMetadata: "meta-4-1",
120144
CommittedLeaderEpoch: 11,
121145
},
122146
},
@@ -129,24 +153,24 @@ func TestOffsetCommitRequest(t *testing.T) {
129153
// GroupInstanceID
130154
for _, version := range []int16{7} {
131155
prototest.TestRequest(t, version, &offsetcommit.Request{
132-
GroupID: "group-4",
156+
GroupID: "group-5",
133157
GenerationID: 1,
134-
MemberID: "member-4",
135-
GroupInstanceID: "instance-4",
158+
MemberID: "member-5",
159+
GroupInstanceID: "instance-5",
136160
Topics: []offsetcommit.RequestTopic{
137161
{
138162
Name: "topic-4",
139163
Partitions: []offsetcommit.RequestPartition{
140164
{
141165
PartitionIndex: 0,
142166
CommittedOffset: 1,
143-
CommittedMetadata: "meta-4-0",
167+
CommittedMetadata: "meta-5-0",
144168
CommittedLeaderEpoch: 10,
145169
},
146170
{
147171
PartitionIndex: 1,
148172
CommittedOffset: 2,
149-
CommittedMetadata: "meta-4-1",
173+
CommittedMetadata: "meta-5-1",
150174
CommittedLeaderEpoch: 11,
151175
},
152176
},

0 commit comments

Comments
 (0)