KAFKA-17019: Producer TimeoutException should include root cause

[chickenchickenlove]

Changes

• Add new Exception class PotentialCauseException.
• All org.apache.kafka.common.errors.TimeoutException in
  KafkaProducer has PotentialCauseException as root cause if it
  cannot catch any exception.

Describe

TimeoutException can be thrown for various reasons.
However, it is often difficult to identify the root cause,
Because there are so many potential factors that can lead to a
TimeoutException.

For example:

1. The ProducerClient might be busy, so it may not be able to send the
   request in time. As a result, some batches may expire, leading to a
   TimeoutException.
2. The broker might be unavailable due to network issues or internal
   failures.
3. A request may be in flight, and although the broker successfully
   handles and responds to it, the response might arrive slightly late.

As shown above, there are many possible causes. In some cases, no
exception is caught in the catch block, and a TimeoutException is
thrown simply by comparing the elapsed time. However, the developer
using TimeoutException in KafkaProducer likely already knows which
specific reasons could cause it in that context. Therefore, I think it
would be helpful to include a PotentialCauseException that reflects
the likely reason, based on the developer’s knowledge.

[frankvicky]

@chickenchickenlove: Thanks for the patch.

[frankvicky]

Could we change PotentialCauseException to Supplier<PotentialCauseException>?
Creating an exception instance is very expensive; lazy creation could help us avoid this situation.

[chickenchickenlove]

Thanks for your comments and sounds great!
I have addressed PR review.
When you have time, please take another look. 🙇‍♂️

[AndrewJSchofield]

Thanks for the PR. Because you've created a new public exception class org.apache.kafka.common.errors.PotentialCauseException, I think you'll need a KIP. Coincidentally, I am trying to track down a bogus TimeoutException to fix an issue in the producer :)

[chia7712]

@chickenchickenlove thanks for this patch. one major comment is left. Please take a a look

[chia7712]

It seems PotentialCauseException does not offer more useful information than the error message, right? Perhaps we could enrich the error message instead of introducing a new nonspecific exception

[chickenchickenlove]

@chia7712 thanks for your comment. 🙇‍♂️
I agree your opinion especially that 'we could enrich the error message instead of introducing a new nonspecific exception'.
However, Introducing a new nonspecific exception to solve this issue has has pros and cons.

Let me explain more.

The KAFKA-17019 requires that all org.apache.kafka.common.errors.TimeoutException should include root cause as a nested exception.

It’s hard to pinpoint the root cause in every situation where a TimeoutException occurs

kafka/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

Lines 405 to 418 in c6cf517

	accumulator.resetNextBatchExpiryTime();
	List<ProducerBatch> expiredInflightBatches = getExpiredInflightBatches(now);
	List<ProducerBatch> expiredBatches = this.accumulator.expiredBatches(now);
	expiredBatches.addAll(expiredInflightBatches);
	// Reset the producer id if an expired batch has previously been sent to the broker. Also update the metrics
	// for expired batches. see the documentation of @TransactionState.resetIdempotentProducerId to understand why
	// we need to reset the producer id here.
	if (!expiredBatches.isEmpty())
	log.trace("Expired {} batches in accumulator", expiredBatches.size());
	for (ProducerBatch expiredBatch : expiredBatches) {
	String errorMessage = "Expiring " + expiredBatch.recordCount + " record(s) for " + expiredBatch.topicPartition
	+ ":" + (now - expiredBatch.createdMs) + " ms has passed since batch creation";
	failBatch(expiredBatch, new TimeoutException(errorMessage), false);

In here, expiredInflightBatches and expiredBatches can cause TimeoutException.
However, TimeoutException is thrown as a result of calculating elapsed time.
Therefore, expiredInflightBatches and expiredBatches don't have information about root cause.

If expiredInflight encounters some problem --

1. Normal Case - No errors and succeed.
2. Network Issue - Error might occurs after the expired period.
3. a little bit slow network issue because of Bandwidth, and so on. - No errors and slow response.
4. Busy CPU due to various reasons - No errors and called slowly.

There are many possible scenarios where a TimeoutException can occur, even with a simple analysis.
However, the expiredInFlight instance can actually throw an error in only 2 (Maybe connection closed, fail to establish connection, ...)

After spending a considerable amount of time reviewing the code and analyzing the TimeoutExceptions thrown by the Producer, I concluded that it's difficult to extract the root cause at every point where a TimeoutException is created.

Idea + Pros and Cons

Developers usually have a good understanding of what kind of error might occur in a given context.
Therefore, in cases where it's difficult to catch the actual root cause, it's possible to include an expected exception as the root cause instead.

Here’s a summary of the pros and cons (compared to simply enhancing the error message):

• pros : At call sites(For example, kafka streams, kafka connect, kafka producer internal and so on) where a TimeoutException is expected, the root cause can be used to handle different cases conditionally.
  • For example, you could create NetworkPotentialCauseException and CPUBusyPotentialCauseException as subclasses of PotentialCauseException, and handle different branches based on the root cause—branch A if the root cause is a NetworkPotentialCauseException, and branch B if it’s a CPUBusyPotentialCauseException.
• cons : Higher instance creation cost compared to String instance.

I spent quite a bit of time thinking through the direction of this PR.
What are your thoughts on it?
Please let me know. 🙇‍♂️

[chia7712]

thanks for your explanation. I agree that specific exception types could help developer to catch the actual root cause. My question is, do we really need PotentialCauseException to be the base exception for NetworkPotentialCauseException and CPUBusyPotentialCauseException? Perhaps KafkaException is good enough.

Another exception design uses TimeoutException as a parent class, similar to e032a36. The benefit of this is simplifying the code, since developers wouldn't need to check root cause of a TimeoutException from root cause of an ExecutionException 😄

thanks for bringing up this great discussion. I'd like to see kafka exception hierarchy becomr more developer-friendly.

[chickenchickenlove]

@chia7712
Thank you for considering my proposal and for sharing your valuable thoughts.
As you mentioned, I also think it’s a good idea to reuse an existing Exception class.

Personally, I think using KafkaException would be a better approach.
Since there can be multiple potential causes for a TimeoutException in certain code paths, it might be difficult to pinpoint the exact cause. In such cases, it could be unclear which subclass of TimeoutException should be used. (e032a36)

So, my suggestion is as follows:

• Include a KafkaException as the root cause of the TimeoutException, and describe the possible scenario in the error message of the KafkaException.
• Let the detailed error information be available via the root cause of the TimeoutException.
• Keep the current message format of the TimeoutException itself (which currently only includes the elapsed time before it expired).

I’m thinking of revising the PR in this direction.

This way, I believe we can preserve the current semantics of TimeoutException while still conveying helpful contextual information (such as a potential cause) when necessary.
In the future, if there's a need to branch logic based on a more specific cause of the timeout—even if no actual exception was thrown—then the developer could define a concrete PotentialCauseException class as needed.

Also, if this direction sounds reasonable, I think it wouldn’t require a KIP change.

What do you think?
Please share your opinion 🙇‍♂️

[frankvicky]

The proposal of reusing KafkaException sounds good. +1 to it.

[AndrewJSchofield]

Sounds good to me also. There's no need to create a KIP for this approach and it gives just as much information to the user.

[chickenchickenlove]

@AndrewJSchofield Thanks for your comments 🙇‍♂️
I'm new to make a KIP.
I have one question!

https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Improvement+Proposals
Should I just go ahead and create a KIP document here?
Once it’s created, will other committers, contributors, and users join the discussion by leaving comments, and then we can proceed with the discussion from there?

Thanks in advance!

[AndrewJSchofield]

@AndrewJSchofield Thanks for your comments 🙇‍♂️ I'm new to make a KIP. I have one question!

https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Improvement+Proposals Should I just go ahead and create a KIP document here? Once it’s created, will other committers, contributors, and users join the discussion by leaving comments, and then we can proceed with the discussion from there?

Thanks in advance!

Yes, that's right. You would follow the instructions in that wiki page and make a new KIP using the template. The discussion would proceed on the dev mailing list. For an interesting KIP, the community will generally not need any encouragement to get involved in the discussion.

[chickenchickenlove]

@chia7712 @AndrewJSchofield @frankvicky Hi!
I made a new commit to address PR review.
When you have time, Please take another look. 🙇‍♂️