KAFKA-15562: ensure commit request manager handles errors correctly

[philipnee]

The current commitRequestManager lacks logic handling various failures. In the patch I addressed the following gap:

1. Network disconnection should disconnect the coordinator node
2. Handling and testing various of fatal and retriable errors scenarios
3. Ensure the time returned in the poll results is Long.MAX or the minimum of all the inflight request's remainingBackoffMs. Because we need to respect the backoff.

[lucasbru]

Thanks @philipnee, I left some comments

[philipnee]

Hi @lucasbru - Thanks for taking the time to review my PR. I addressed all but 2 comments:

1. Mockito: Could you be more specific on how you expect to mock the response object?
2. Error handling: Essentially all errors in continueHandlePartitionErrors can only happen in the response. I understand there are some redundancy there and can be a bit confusing. But the response and both throw a hard failures (response = null and throwable = non-null) or a server side error (response = non-null). That is why it was kept separated. If you find it unclear - how do I make it more readable?

[lucasbru]

nit: merge first two branches now?

[lucasbru]

Changes look good to me. Did I understand david correctly that we want to add more changes to this PR?

1. Mockito: Could you be more specific on how you expect to mock the response object?

That was just "out of interest", since you are already using mockito to mock things, why not use it for the response object as well. But I realize that the object is a pretty plain data container, so the answer I guess is that it's easier.

2. Error handling: Essentially all errors in continueHandlePartitionErrors can only happen in the response. I understand there are some redundancy there and can be a bit confusing. But the response and both throw a hard failures (response = null and throwable = non-null) or a server side error (response = non-null). That is why it was kept separated. If you find it unclear - how do I make it more readable?

That's not what I meant. I just found the flow onResponse a bit hard to read, so I suggested structuring it a bit differently. In particular, continueHandlePartitionErrors sounds like that it's determining whether to continue handling errors, but it's actually completing the future inside.

I think the control flow may also lead to logging in a way that may not be intended? For fencing, we get one generic log.error message and one very similar log.info message with some specific details. For topic authorization errors, we get one log.error message for each partition, and then another aggregated log.error message, listing the partitions again. However, if we have 3 unauthorized topic errors and then one other error, we get three generic error messages for the topics, but we do not get the aggregated error message. Is that trying to be consistent with the old consumer?

I imagined something like:

if (error == Errors.NONE) {
    log.debug("OffsetCommit {} for partition {}", offset, tp);
} else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
    // Collect all unauthorized topics before failing
    unauthorizedTopics.add(tp.topic());
} else if (error instanceof RetriableException) {
    log.warn("OffsetCommit failed on partition {} at offset {}: {}", tp, offset, error.message());
    handleRetriableError(error);
    retry(responseTime);
    return;
} else {
    log.error("OffsetCommit failed on partition {} at offset {}: {}", tp, offset, error.message());
    handleFatalError(error);
    return;
}

But I'm not super familiar with the code style in the new consumer, and consistency with the rest of the code and the old consumer is also important. So I just wanted to give this to you as an inspiration, but you are probably in the best position to come up with the best way to implement it.

[philipnee]

Hi @lucasbru - Thanks for the response. To your first question: Yes, what is left is correcting the error handling at the user API level. At the time I implemented this, I wrapped all errors in KafkaException (I forgot the reason why), however, we should throw the exact exception instead. So I will follow up with a second part PR to make this Jira issue completed.

I get what you meant for onResponse and I think your suggestion makes a lot of sense. I'll patch it accordingly.

[lucasbru]

I think we're having duplicate logging now.

[lucasbru]

I think you want to remove this if/else block completely now

[philipnee]

Thanks, as well as the error == Errors.NONE

[philipnee]

@lucasbru - Thanks for taking time reviewing this PR. I addressed your previous comment. Let me know if there's anything unclear on this PR.

[philipnee]

Test failures seem irrelevant, but retriggering the test as there's a failed build

[lucasbru]

LGTM. Thanks, @philipnee !

[lucasbru]

Test failures unrelated