Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(proto): use full range of MetadataRequest #2556

Merged
merged 1 commit into from
Aug 4, 2023
Merged

fix(proto): use full range of MetadataRequest #2556

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 11 additions & 11 deletions async_producer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1111,7 +1111,7 @@ func TestAsyncProducerIdempotentGoldenPath(t *testing.T) {
broker := NewMockBroker(t, 1)

metadataResponse := &MetadataResponse{
Version: 1,
Version: 4,
ControllerID: 1,
}
metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
Expand Down Expand Up @@ -1169,7 +1169,7 @@ func TestAsyncProducerIdempotentRetryCheckBatch(t *testing.T) {
broker := NewMockBroker(t, 1)

metadataResponse := &MetadataResponse{
Version: 1,
Version: 4,
ControllerID: 1,
}
metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
Expand Down Expand Up @@ -1307,7 +1307,7 @@ func TestAsyncProducerIdempotentRetryCheckBatch_2378(t *testing.T) {
broker := NewMockBroker(t, 1)

metadataResponse := &MetadataResponse{
Version: 1,
Version: 4,
ControllerID: 1,
}
metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
Expand Down Expand Up @@ -1375,7 +1375,7 @@ func TestAsyncProducerIdempotentErrorOnOutOfSeq(t *testing.T) {
broker := NewMockBroker(t, 1)

metadataResponse := &MetadataResponse{
Version: 1,
Version: 4,
ControllerID: 1,
}
metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
Expand Down Expand Up @@ -1425,7 +1425,7 @@ func TestAsyncProducerIdempotentEpochRollover(t *testing.T) {
defer broker.Close()

metadataResponse := &MetadataResponse{
Version: 1,
Version: 4,
ControllerID: 1,
}
metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
Expand Down Expand Up @@ -1501,7 +1501,7 @@ func TestAsyncProducerIdempotentEpochExhaustion(t *testing.T) {
)

metadataResponse := &MetadataResponse{
Version: 1,
Version: 4,
ControllerID: 1,
}
metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
Expand Down Expand Up @@ -1737,7 +1737,7 @@ func TestTxmngInitProducerId(t *testing.T) {
defer broker.Close()

metadataLeader := new(MetadataResponse)
metadataLeader.Version = 1
metadataLeader.Version = 4
metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
broker.Returns(metadataLeader)

Expand Down Expand Up @@ -1879,7 +1879,7 @@ func TestTxnProduceRecordWithCommit(t *testing.T) {
config.Net.MaxOpenRequests = 1

metadataLeader := new(MetadataResponse)
metadataLeader.Version = 1
metadataLeader.Version = 4
metadataLeader.ControllerID = broker.brokerID
metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
metadataLeader.AddTopic("test-topic", ErrNoError)
Expand Down Expand Up @@ -1960,7 +1960,7 @@ func TestTxnProduceBatchAddPartition(t *testing.T) {
config.Producer.Partitioner = NewManualPartitioner

metadataLeader := new(MetadataResponse)
metadataLeader.Version = 1
metadataLeader.Version = 4
metadataLeader.ControllerID = broker.brokerID
metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
metadataLeader.AddTopic("test-topic", ErrNoError)
Expand Down Expand Up @@ -2067,7 +2067,7 @@ func TestTxnProduceRecordWithAbort(t *testing.T) {
config.Net.MaxOpenRequests = 1

metadataLeader := new(MetadataResponse)
metadataLeader.Version = 1
metadataLeader.Version = 4
metadataLeader.ControllerID = broker.brokerID
metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
metadataLeader.AddTopic("test-topic", ErrNoError)
Expand Down Expand Up @@ -2147,7 +2147,7 @@ func TestTxnCanAbort(t *testing.T) {
config.Net.MaxOpenRequests = 1

metadataLeader := new(MetadataResponse)
metadataLeader.Version = 1
metadataLeader.Version = 4
metadataLeader.ControllerID = broker.brokerID
metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
metadataLeader.AddTopic("test-topic", ErrNoError)
Expand Down
2 changes: 1 addition & 1 deletion client_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1071,7 +1071,7 @@ func TestClientCoordinatorConnectionRefused(t *testing.T) {
func TestInitProducerIDConnectionRefused(t *testing.T) {
t.Parallel()
seedBroker := NewMockBroker(t, 1)
seedBroker.Returns(&MetadataResponse{Version: 1})
seedBroker.Returns(&MetadataResponse{Version: 4})

config := NewTestConfig()
config.Producer.Idempotent = true
Expand Down
28 changes: 17 additions & 11 deletions metadata_request.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,10 @@ func NewMetadataRequest(version KafkaVersion, topics []string) *MetadataRequest
m.Version = 6
} else if version.IsAtLeast(V1_0_0_0) {
m.Version = 5
} else if version.IsAtLeast(V0_11_0_0) {
Copy link
Collaborator

@hindessm hindessm Aug 4, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

FYI: I prefer these else if highest first which makes me dislike those if {}; if {} lowest first additions I queried yesterday even more.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@hindessm I agree it's nicer and it works here because there's no mandatory additional fields, just the version number being set.

However, In some of those other cases it (e.g.,) needs a mandatory field included from V4 upward, so an additive approach works better.

One additional thing I did like about Metadata was that it puts the version choice logic alongside the proto definition rather than within client.go / broker.go / consumer.go etc.

m.Version = 4
} else if version.IsAtLeast(V0_10_1_0) {
m.Version = 2
} else if version.IsAtLeast(V0_10_0_0) {
m.Version = 1
}
Expand Down Expand Up @@ -94,19 +98,21 @@ func (r *MetadataRequest) isValidVersion() bool {

func (r *MetadataRequest) requiredVersion() KafkaVersion {
switch r.Version {
case 1:
return V0_10_0_0
case 2:
return V0_10_1_0
case 3, 4:
return V0_11_0_0
case 5:
return V1_0_0_0
case 6:
return V2_0_0_0
case 7:
return V2_1_0_0
case 6:
return V2_0_0_0
case 5:
return V1_0_0_0
case 3, 4:
return V0_11_0_0
case 2:
return V0_10_1_0
case 1:
return V0_10_0_0
case 0:
return V0_8_2_0
default:
return MinVersion
return V2_1_0_0
}
}
24 changes: 13 additions & 11 deletions metadata_response.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -281,20 +281,22 @@ func (r *MetadataResponse) isValidVersion() bool {

func (r *MetadataResponse) requiredVersion() KafkaVersion {
switch r.Version {
case 1:
return V0_10_0_0
case 2:
return V0_10_1_0
case 3, 4:
return V0_11_0_0
case 5:
return V1_0_0_0
case 6:
return V2_0_0_0
case 7:
return V2_1_0_0
case 6:
return V2_0_0_0
case 5:
return V1_0_0_0
case 3, 4:
return V0_11_0_0
case 2:
return V0_10_1_0
case 1:
return V0_10_0_0
case 0:
return V0_8_2_0
default:
return MinVersion
return V2_1_0_0
}
}

Expand Down
2 changes: 1 addition & 1 deletion sync_producer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ func TestSyncProducerTransactional(t *testing.T) {
config.Net.MaxOpenRequests = 1

metadataResponse := new(MetadataResponse)
metadataResponse.Version = 1
metadataResponse.Version = 4
metadataResponse.ControllerID = leader.BrokerID()
metadataResponse.AddBroker(leader.Addr(), leader.BrokerID())
metadataResponse.AddTopic("my_topic", ErrNoError)
Expand Down
12 changes: 6 additions & 6 deletions transaction_manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ func TestTxnmgrInitProducerIdTxn(t *testing.T) {
defer broker.Close()

metadataLeader := new(MetadataResponse)
metadataLeader.Version = 1
metadataLeader.Version = 4
metadataLeader.ControllerID = broker.brokerID
metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
broker.Returns(metadataLeader)
Expand Down Expand Up @@ -217,7 +217,7 @@ func TestMaybeAddPartitionToCurrentTxn(t *testing.T) {
defer broker.Close()

metadataLeader := new(MetadataResponse)
metadataLeader.Version = 1
metadataLeader.Version = 4
metadataLeader.ControllerID = broker.brokerID
metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
metadataLeader.AddTopic("test-topic", ErrNoError)
Expand Down Expand Up @@ -351,7 +351,7 @@ func TestAddOffsetsToTxn(t *testing.T) {
defer broker.Close()

metadataLeader := new(MetadataResponse)
metadataLeader.Version = 1
metadataLeader.Version = 4
metadataLeader.ControllerID = broker.brokerID
metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
metadataLeader.AddTopic("test-topic", ErrNoError)
Expand Down Expand Up @@ -608,7 +608,7 @@ func TestTxnOffsetsCommit(t *testing.T) {
config.Producer.Transaction.Retry.Backoff = 0

metadataLeader := new(MetadataResponse)
metadataLeader.Version = 1
metadataLeader.Version = 4
metadataLeader.ControllerID = broker.brokerID
metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
metadataLeader.AddTopic("test-topic", ErrNoError)
Expand Down Expand Up @@ -739,7 +739,7 @@ func TestEndTxn(t *testing.T) {
defer broker.Close()

metadataLeader := new(MetadataResponse)
metadataLeader.Version = 1
metadataLeader.Version = 4
metadataLeader.ControllerID = broker.brokerID
metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
metadataLeader.AddTopic("test-topic", ErrNoError)
Expand Down Expand Up @@ -904,7 +904,7 @@ func TestPublishPartitionToTxn(t *testing.T) {
defer broker.Close()

metadataLeader := new(MetadataResponse)
metadataLeader.Version = 1
metadataLeader.Version = 4
metadataLeader.ControllerID = broker.brokerID
metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
metadataLeader.AddTopic("test-topic", ErrNoError)
Expand Down