MINOR: Improve EOS example exception handling

[abbccdda]

The current EOS example is over-complicating the exception handling by mixing non fatal and fatal ones. This cleanup is trying to make the code readability better. Will rebase after #8051 is merged

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[hachikuji]

I don't think this is the only case that we can abort a transaction. This only handles offset commit failures, but what about send failures?

[hachikuji]

Thanks for updating this. Left a few comments.

[hachikuji]

Why is InvalidOffsetException fatal?

[abbccdda]

I guess it could be retriable.

[hachikuji]

Could we do "group mode" only in this example? The example doesn't really extend to multiple instances otherwise.

[abbccdda]

Not sure I follow, we could manually compute the partitions to assign to for the standalone mode?

[hachikuji]

Hmm.. Is it actually safe to abort following a TimeoutException? I think this would cause an illegal state error in the producer. To handle this correctly, the application should retry the commit.

[abbccdda]

Or we shouldn't handle the timeout, as we always rely on max_block to ensure the request is successful?

[hachikuji]

For the purpose of understanding EOS, the main exceptions that are worth calling out are ProducerFencedException and FencedInstanceIdException. I would suggest we write the example like this:

try {
   ...
   producer.commitTransaction;
} catch (ProducerFencedException e) {
  throw KafkaException("The transactional.id $transactionalId has been claimed by another process");
} catch (FencedInstanceIdException e) {
  throw KafkaException("The group.instance.id $instanceId has been claimed by another process");
} catch (KafkaException e) {
  // If we have not been fenced, try to abort the transaction and continue. This will raise immediately
  // if the producer has hit a fatal error.
  producer.abortTransaction();
}

[hachikuji]

retest this please

[hachikuji]

LGTM. Thanks for simplifying this example.

[hachikuji]

retest this please

[hachikuji]

ok to test

[guozhangwang]

Cherry-picked to 2.5