Retries

[jroland]

Retries

[Jroland]

I agree Timothy, this is an area of the driver that has zero tests and is not well flushed out. There is a little TODO comment in the BrokerRouter class which is basically saying this. This is the next bit that I am working on. I appreciate the comment, let me know if you have any other thoughts on areas to concentrate on.

[Jroland]

Here is the relevant documentation which describes this:
The client does not need to keep polling to see if the cluster has changed; it can fetch metadata once when it is instantiated cache that metadata until it receives an error indicating that the metadata is out of date. This error can come in two forms: (1) a socket error indicating the client cannot communicate with a particular broker, (2) an error code in the response to a request indicating that this broker no longer hosts the partition for which data was requested.

[Jroland]

Also the documentation states:
https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-MetadataResponse

Leader
The node id for the kafka broker currently acting as leader for this partition. If no leader exists because we are in the middle of a leader election this id will be -1.

We need to handle retries against all these cases listed.

[Jroland]

Relevant Error Codes
LeaderNotAvailable
5
This error is thrown if we are in the middle of a leadership election and there is currently no leader for this partition and hence it is unavailable for writes.

NotLeaderForPartition
6
This error is thrown if the client attempts to send messages to a replica that is not the leader for some partition. It indicates that the clients metadata is out of date.

And possibly
ConsumerCoordinatorNotAvailableCode 15 The broker returns this error code for consumer metadata requests or offset commit requests if the offsets topic has not yet been created.

[tnachen]

Yes you need to handle soft failures like LeaderNotAvailabe, etc. There are hard failures too such as MessageTooLarge or UnknownTopicAndPartition that retry cannot recover at all. So clients should only retry on soft failures. Currently we do exponential backoff on retries too to avoid hammering the brokers too much especially on timeouts since it can hammer the brokers too much and cause slow perf.

[vchekan]

Just a note. "LeaderNotAvailable" is returned for auto-created topics.
https://github.com/apache/kafka/blob/0.8.1/core/src/main/scala/kafka/server/KafkaApis.scala#L664
So there it is a bug to accept all responses as successful ones in kafka-net:
https://github.com/Jroland/kafka-net/blob/master/src/kafka-net/BrokerRouter.cs#L251
Only those with error=0 should be accepted:
https://github.com/apache/kafka/blob/0.8.1/core/src/main/scala/kafka/common/ErrorMapping.scala#L32

[vchekan]

Hi @Jroland,
Have you started working on this issue? If not yet, I'd like to give it a try.

Here are requirements I'd like to satisfy.

Requirements:

1. Driver should distinguish intermittent errors from permanent ones.
2. Driver should attempt to recover from intermittent errors.
3. Recovery should be limited by timeout.

API:
4. Client should be notified that intermittent error happen.
5. Client should be notified that permanent error happen.
6. Client should be notified that intermittent error timed out and becomes permanent one.
7. When error notification happen, client should be provided with messages which have failed.
8. Upon shutdown, a flush of internal buffer should be attempted. Flush time should be limited with timeout.
9. After flush attempt, If there are messages left in the buffer, client should be notified and list of messages should be provided.

What do you think?

[tnachen]

Also one thing to consider is that some failures (and probably most of the soft failures) implies a Metadata update is required.

[Jroland]

Hey @vchekan, the only thing I have started to do with this issue is bubble up the disconnection event from the kafka socket class so the BrokerRouter is aware that we have lost connection to a broker. Otherwise I do not have any of the logic implemented for gracefully handling metadata changes.

So go ahead and start putting something together for handling this. It is the most urgent issue for the driver at the moment by far. My advice to you here on this one is start small and implement it in testable sections. It might be hard to pull it into the main branch otherwise.