kafka new producer is not able to update metadata after one of the broker is down

Question

I have an kafka environment which has 2 brokers and 1 zookeeper.

While I am trying to produce messages to kafka, if i stop broker 1(which is the leader one) the client stops producing messaging and give me the below error although the broker 2 is elected as a new leader for the topic and partions.

org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.

After 10 minutes passed, since broker 2 is new leader i expected producer to send data to broker 2 but it continued failing by giving above exception. lastRefreshMs and lastSuccessfullRefreshMs is still same although the metadataExpireMs is 300000 for producer.

I am using kafka new Producer implementation on producer side.

It seems that when producer is initiated, it binds to one broker and if that broker goes down it is not even trying to connect to another brokers in cluster.

But my expectation is if a broker goes down, it should directly check metadata for another brokers that are available and send data to them.

Btw my topic is 4 partition and has replication factor of 2. Giving this info in case it makes sense.

Configuration params.

{request.timeout.ms=30000, retry.backoff.ms=100, buffer.memory=33554432, ssl.truststore.password=null, batch.size=16384, ssl.keymanager.algorithm=SunX509, receive.buffer.bytes=32768, ssl.cipher.suites=null, ssl.key.password=null, sasl.kerberos.ticket.renew.jitter=0.05, ssl.provider=null, sasl.kerberos.service.name=null, max.in.flight.requests.per.connection=5, sasl.kerberos.ticket.renew.window.factor=0.8, bootstrap.servers=[10.201.83.166:9500, 10.201.83.167:9500], client.id=rest-interface, max.request.size=1048576, acks=1, linger.ms=0, sasl.kerberos.kinit.cmd=/usr/bin/kinit, ssl.enabled.protocols=[TLSv1.2, TLSv1.1, TLSv1], metadata.fetch.timeout.ms=60000, ssl.endpoint.identification.algorithm=null, ssl.keystore.location=null, value.serializer=class org.apache.kafka.common.serialization.ByteArraySerializer, ssl.truststore.location=null, ssl.keystore.password=null, key.serializer=class org.apache.kafka.common.serialization.ByteArraySerializer, block.on.buffer.full=false, metrics.sample.window.ms=30000, metadata.max.age.ms=300000, security.protocol=PLAINTEXT, ssl.protocol=TLS, sasl.kerberos.min.time.before.relogin=60000, timeout.ms=30000, connections.max.idle.ms=540000, ssl.trustmanager.algorithm=PKIX, metric.reporters=[], compression.type=none, ssl.truststore.type=JKS, max.block.ms=60000, retries=0, send.buffer.bytes=131072, partitioner.class=class org.apache.kafka.clients.producer.internals.DefaultPartitioner, reconnect.backoff.ms=50, metrics.num.samples=2, ssl.keystore.type=JKS}

Use Case:

1- Start BR1 and BR2 Produce data (Leader is BR1)

2- Stop BR2 produce data(fine)

3- Stop BR1(which means there is no active working broker in cluster at this time) and then Start BR2 and produce data (failed although leader is BR2)

4- Start BR1 produce data(leader is still BR2 but data is produced finely)

5- Stop BR2(now BR1 is leader)

6- Stop BR1(BR1 is still leader)

7- Start BR1 produce data(message is produced fine again)

If producer send the latest successful data to BR1 and then all brokers goes down, the producer expects BR1 to get up again although BR2 is up and new leader. Is this an expected behaviour?

Solution 1

After spending hours I figured out the behaviour of kafka in my situation. May be this is a bug or may be this needs to be done this way for the reasons lie under the hood but actually if i would do such implementation i wouldn't do this way :)

When all brokers goes down, if you are able to get up only one broker this must be the broker which went down last in order to produce messages successfully.

Let's say you have 5 brokers; BR1, BR2, BR3, BR4 and BR5. If all goes down and if the lastly dead broker is BR3(which was the last leader), although you start all brokers BR1, BR2, BR4 and BR5, it will not make any sense unless you start BR3.

Solution 2

You need to increase the number of retries. In your case you need to set it to >=5.

That is the only way for your producer to know that your cluster has a new leader.

Besides that, make sure that all your brokers have a copy of your partition(s). Else you aren't going to get a new leader.

Comments

[Hans Jespersen]

I believe the behavior depends on the broker settings for unclean.leader.election.enable and min.insync.replicas. If an unclean leader election is allowed then one of the other brokers can become the new leader and you should be able to publish even with the last broker is still down. See this presentation for all the tradeoffs slideshare.net/gwenshap/…

[John]

Hi jit, although I tried @HansJespersen suggestion , I have the same situation as you faced. Have you find out any solution to this ? Or this is how kafka works ?

[jit]

@John Hans is right but this time you trade-off consistency with availability. Are you sure you passed the parameters correctly as Hans mentioned. This parameter names may also vary for different kafka versions. Look for the correct one for your version.

[John]

Test Configuration is : 3 broker 1 topic 1 partition 3 replication factor. unclean.leader.election.enable=true and min.insync.replicas 1. closing brokers one-by-one. Using Kafka 2.12-2.1.1. Not working. Still trying to connect latest connected broker. What could be causing this ?

[jit]

@John on the producer side there is a config parameter named metadata.max.age.ms. Set this value as low as you can to detect newly selected leader.