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(client): ignore empty Metadata responses when refreshing #2672

Merged
merged 1 commit into from
Oct 13, 2023
Merged

fix(client): ignore empty Metadata responses when refreshing #2672

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
7 changes: 7 additions & 0 deletions client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1035,6 +1035,13 @@ func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int,
var kerror KError
var packetEncodingError PacketEncodingError
if err == nil {
// When talking to the startup phase of a broker, it is possible to receive an empty metadata set. We should remove that broker and try next broker (https://issues.apache.org/jira/browse/KAFKA-7924).
if len(response.Brokers) == 0 {
Logger.Println("client/metadata receiving empty brokers from the metadata response when requesting the broker #%d at %s", broker.ID(), broker.addr)
_ = broker.Close()
client.deregisterBroker(broker)
continue
}
allKnownMetaData := len(topics) == 0
// valid response, use it
shouldRetry, err := client.updateMetadata(response, allKnownMetaData)
Expand Down
86 changes: 78 additions & 8 deletions client_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,9 @@ func safeClose(t testing.TB, c io.Closer) {
func TestSimpleClient(t *testing.T) {
seedBroker := NewMockBroker(t, 1)

seedBroker.Returns(new(MetadataResponse))
metadataResponse := new(MetadataResponse)
metadataResponse.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
seedBroker.Returns(metadataResponse)

client, err := NewClient([]string{seedBroker.Addr()}, NewTestConfig())
if err != nil {
Expand Down Expand Up @@ -92,6 +94,7 @@ func TestClientDoesntCachePartitionsForTopicsWithErrors(t *testing.T) {
}

metadataResponse = new(MetadataResponse)
metadataResponse.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
metadataResponse.AddTopic("unknown", ErrUnknownTopicOrPartition)
seedBroker.Returns(metadataResponse)

Expand All @@ -111,6 +114,7 @@ func TestClientDoesntCachePartitionsForTopicsWithErrors(t *testing.T) {
}

metadataResponse = new(MetadataResponse)
metadataResponse.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
metadataResponse.AddTopic("unknown", ErrUnknownTopicOrPartition)
seedBroker.Returns(metadataResponse)

Expand Down Expand Up @@ -358,6 +362,7 @@ func TestClientReceivingUnknownTopicWithBackoffFunc(t *testing.T) {
seedBroker := NewMockBroker(t, 1)

metadataResponse1 := new(MetadataResponse)
metadataResponse1.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
seedBroker.Returns(metadataResponse1)

retryCount := int32(0)
Expand All @@ -375,6 +380,7 @@ func TestClientReceivingUnknownTopicWithBackoffFunc(t *testing.T) {

metadataUnknownTopic := new(MetadataResponse)
metadataUnknownTopic.AddTopic("new_topic", ErrUnknownTopicOrPartition)
metadataUnknownTopic.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
seedBroker.Returns(metadataUnknownTopic)
seedBroker.Returns(metadataUnknownTopic)

Expand All @@ -395,6 +401,7 @@ func TestClientReceivingUnknownTopic(t *testing.T) {
seedBroker := NewMockBroker(t, 1)

metadataResponse1 := new(MetadataResponse)
metadataResponse1.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
seedBroker.Returns(metadataResponse1)

config := NewTestConfig()
Expand All @@ -406,6 +413,7 @@ func TestClientReceivingUnknownTopic(t *testing.T) {
}

metadataUnknownTopic := new(MetadataResponse)
metadataUnknownTopic.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
metadataUnknownTopic.AddTopic("new_topic", ErrUnknownTopicOrPartition)
seedBroker.Returns(metadataUnknownTopic)
seedBroker.Returns(metadataUnknownTopic)
Expand Down Expand Up @@ -481,6 +489,53 @@ func TestClientReceivingPartialMetadata(t *testing.T) {
leader.Close()
}

func TestClientRefreshBehaviourWhenEmptyMetadataResponse(t *testing.T) {
seedBroker := NewMockBroker(t, 1)
broker := NewMockBroker(t, 2)

metadataResponse1 := new(MetadataResponse)
metadataResponse1.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
seedBroker.Returns(metadataResponse1)

c, err := NewClient([]string{seedBroker.Addr()}, NewTestConfig())
if err != nil {
t.Fatal(err)
}
client := c.(*client)
if len(client.seedBrokers) != 1 {
t.Error("incorrect number of live seeds")
}
if len(client.deadSeeds) != 0 {
t.Error("incorrect number of dead seeds")
}
if len(client.brokers) != 1 {
t.Error("incorrect number of brokers")
}

// Empty metadata response
seedBroker.Returns(new(MetadataResponse))
metadataResponse2 := new(MetadataResponse)
metadataResponse2.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
metadataResponse2.AddBroker(broker.Addr(), broker.BrokerID())
seedBroker.Returns(metadataResponse2)
err = c.RefreshMetadata()
if err != nil {
t.Fatal(err)
}
if len(client.seedBrokers) != 1 {
t.Error("incorrect number of live seeds")
}
if len(client.deadSeeds) != 0 {
t.Error("incorrect number of dead seeds")
}
if len(client.brokers) != 2 {
t.Error("incorrect number of brokers")
}
broker.Close()
seedBroker.Close()
safeClose(t, client)
}

func TestClientRefreshBehaviour(t *testing.T) {
seedBroker := NewMockBroker(t, 1)
leader := NewMockBroker(t, 5)
Expand Down Expand Up @@ -633,8 +688,9 @@ func TestClientGetBroker(t *testing.T) {

func TestClientResurrectDeadSeeds(t *testing.T) {
initialSeed := NewMockBroker(t, 0)
emptyMetadata := new(MetadataResponse)
initialSeed.Returns(emptyMetadata)
metadataResponse := new(MetadataResponse)
metadataResponse.AddBroker(initialSeed.Addr(), initialSeed.BrokerID())
initialSeed.Returns(metadataResponse)

conf := NewTestConfig()
conf.Metadata.Retry.Backoff = 0
Expand All @@ -643,7 +699,6 @@ func TestClientResurrectDeadSeeds(t *testing.T) {
if err != nil {
t.Fatal(err)
}
initialSeed.Close()

client := c.(*client)

Expand All @@ -658,6 +713,7 @@ func TestClientResurrectDeadSeeds(t *testing.T) {
safeClose(t, client.seedBrokers[0])
client.seedBrokers = []*Broker{NewBroker(addr1), NewBroker(addr2), NewBroker(addr3)}
client.deadSeeds = []*Broker{}
client.brokers = map[int32]*Broker{}

wg := sync.WaitGroup{}
wg.Add(1)
Expand All @@ -676,7 +732,9 @@ func TestClientResurrectDeadSeeds(t *testing.T) {
seed3.Close()

seed1.Close()
seed2.Returns(emptyMetadata)
metadataResponse2 := new(MetadataResponse)
metadataResponse2.AddBroker(seed2.Addr(), seed2.BrokerID())
seed2.Returns(metadataResponse2)

wg.Wait()

Expand Down Expand Up @@ -767,6 +825,7 @@ func TestClientMetadataTimeout(t *testing.T) {
// Use a responsive broker to create a working client
initialSeed := NewMockBroker(t, 0)
emptyMetadata := new(MetadataResponse)
emptyMetadata.AddBroker(initialSeed.Addr(), initialSeed.BrokerID())
initialSeed.Returns(emptyMetadata)

conf := NewTestConfig()
Expand Down Expand Up @@ -996,6 +1055,7 @@ func TestClientCoordinatorWithoutConsumerOffsetsTopic(t *testing.T) {
coordinator := NewMockBroker(t, 2)

metadataResponse1 := new(MetadataResponse)
metadataResponse1.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
seedBroker.Returns(metadataResponse1)

config := NewTestConfig()
Expand All @@ -1011,11 +1071,13 @@ func TestClientCoordinatorWithoutConsumerOffsetsTopic(t *testing.T) {
seedBroker.Returns(coordinatorResponse1)

metadataResponse2 := new(MetadataResponse)
metadataResponse2.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
metadataResponse2.AddTopic("__consumer_offsets", ErrUnknownTopicOrPartition)
seedBroker.Returns(metadataResponse2)

replicas := []int32{coordinator.BrokerID()}
metadataResponse3 := new(MetadataResponse)
metadataResponse3.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
metadataResponse3.AddTopicPartition("__consumer_offsets", 0, replicas[0], replicas, replicas, []int32{}, ErrNoError)
seedBroker.Returns(metadataResponse3)

Expand Down Expand Up @@ -1049,6 +1111,7 @@ func TestClientAutorefreshShutdownRace(t *testing.T) {
defer seedBroker.Close()

metadataResponse := new(MetadataResponse)
metadataResponse.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
seedBroker.Returns(metadataResponse)

conf := NewTestConfig()
Expand Down Expand Up @@ -1105,7 +1168,9 @@ func TestClientConnectionRefused(t *testing.T) {
func TestClientCoordinatorConnectionRefused(t *testing.T) {
t.Parallel()
seedBroker := NewMockBroker(t, 1)
seedBroker.Returns(new(MetadataResponse))
metadataResponse := new(MetadataResponse)
metadataResponse.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
seedBroker.Returns(metadataResponse)

client, err := NewClient([]string{seedBroker.Addr()}, NewTestConfig())
if err != nil {
Expand All @@ -1130,7 +1195,10 @@ func TestClientCoordinatorConnectionRefused(t *testing.T) {
func TestInitProducerIDConnectionRefused(t *testing.T) {
t.Parallel()
seedBroker := NewMockBroker(t, 1)
seedBroker.Returns(&MetadataResponse{Version: 4})
metadataResponse := new(MetadataResponse)
metadataResponse.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
metadataResponse.Version = 4
seedBroker.Returns(metadataResponse)

config := NewTestConfig()
config.Producer.Idempotent = true
Expand Down Expand Up @@ -1161,7 +1229,9 @@ func TestInitProducerIDConnectionRefused(t *testing.T) {
func TestMetricsCleanup(t *testing.T) {
seedBroker := NewMockBroker(t, 1)
defer seedBroker.Close()
seedBroker.Returns(new(MetadataResponse))
metadataResponse := new(MetadataResponse)
metadataResponse.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
seedBroker.Returns(metadataResponse)

config := NewTestConfig()
metrics.GetOrRegisterMeter("a", config.MetricRegistry)
Expand Down
4 changes: 3 additions & 1 deletion client_tls_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -197,7 +197,9 @@ func doListenerTLSTest(t *testing.T, expectSuccess bool, serverConfig, clientCon
seedBroker := NewMockBrokerListener(childT, 1, seedListener)
defer seedBroker.Close()

seedBroker.Returns(new(MetadataResponse))
metadataResponse := new(MetadataResponse)
metadataResponse.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
seedBroker.Returns(metadataResponse)

config := NewTestConfig()
config.Net.TLS.Enable = true
Expand Down
4 changes: 3 additions & 1 deletion offset_manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,9 @@ func initPartitionOffsetManager(t *testing.T, om OffsetManager,

func TestNewOffsetManager(t *testing.T) {
seedBroker := NewMockBroker(t, 1)
seedBroker.Returns(new(MetadataResponse))
metadataResponse := new(MetadataResponse)
metadataResponse.AddBroker(seedBroker.Addr(), seedBroker.BrokerID())
seedBroker.Returns(metadataResponse)
defer seedBroker.Close()

testClient, err := NewClient([]string{seedBroker.Addr()}, NewTestConfig())
Expand Down
1 change: 1 addition & 0 deletions sync_producer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -271,6 +271,7 @@ func TestSyncProducerToNonExistingTopic(t *testing.T) {
}

metadataResponse = new(MetadataResponse)
metadataResponse.AddBroker(broker.Addr(), broker.BrokerID())
metadataResponse.AddTopic("unknown", ErrUnknownTopicOrPartition)
broker.Returns(metadataResponse)

Expand Down
Loading