Add option to specify timeout for how long to wait for blocking Producer#send

[JorgenRingen]

No description provided.

[JorgenRingen]

A little bit unsure regarding providing commit-offset-timeout as an option: https://github.com/confluentinc/parallel-consumer/blob/master/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ConsumerOffsetCommitter.java#L148

What's the reason for why timeouts isn't retried? 🤔

if (take == null)
                    throw new InternalRuntimeError(msg("Timeout waiting for commit response {} to request {}", timeout, commitRequest));

This exception would stop the control-loop.

[astubbs]

A little bit unsure regarding providing commit-offset-timeout as an option: https://github.com/confluentinc/parallel-consumer/blob/master/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ConsumerOffsetCommitter.java#L148

What's the reason for why timeouts isn't retried? 🤔

if (take == null)
                    throw new InternalRuntimeError(msg("Timeout waiting for commit response {} to request {}", timeout, commitRequest));

This exception would stop the control-loop.

It does get retried, but it is retried through the same system as if the user's function threw an error. It was easier to do / simpler. Down side is that the message is of course also reprocessed, so not ideal.

Regarding retrying, do you have any preference for how this should occur? Probably would have it's own PR as this PR as a simple config change is trivial to merge (sorry for delay).

[astubbs]

needs rebasing

[JorgenRingen]

It does get retried, but it is retried through the same system as if the user's function threw an error. It was easier to do / simpler. Down side is that the message is of course also reprocessed, so not ideal.

Regarding retrying, do you have any preference for how this should occur? Probably would have it's own PR as this PR as a simple config change is trivial to merge (sorry for delay).

Perfectly understandable 👍

Agree that reprocessing in user-function should be limited as much as possible, so consumer/producer-errors should preferably be retried in the client lib itself if possible. In this specific case it might just be a matter of aligning the timeout so that the consumer is able to retry exhaustively. Perhaps https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#consumerconfigs_request.timeout.ms is the one 🤔

Duration timeout = ParallelEoSStreamProcessor.DEFAULT_TIMEOUT;
                CommitResponse take = commitResponseQueue.poll(timeout.toMillis(), TimeUnit.MILLISECONDS); // blocks, drain until we find our response

NB: This exception actually caused the control-loop to stop as far as I remember, but msg was reprocessed after application restart :)

[astubbs]

NB: This exception actually caused the control-loop to stop as far as I remember, but msg was reprocessed after application restart :)

Sorry, what exception caused the system to shutdown?

[astubbs]

I think you need to mark this as ready for review?

[JorgenRingen]

NB: This exception actually caused the control-loop to stop as far as I remember, but msg was reprocessed after application restart :)

Sorry, what exception caused the system to shutdown?

If offset-commit timed out (10sec default): https://github.com/confluentinc/parallel-consumer/blob/master/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ConsumerOffsetCommitter.java#L156

I think you need to mark this as ready for review?

Sorry, rebased and marked as ready! Build failed, but not possible to see details: https://jenkins.confluent.io/job/confluentinc-pr/job/parallel-consumer/job/PR-121/3/display/redirect

[astubbs]

Ah sorry, our build system is under maintenance atm.

ERROR: We have disabled builds for pull requests created by the open source communitydue to build system maintenance.We are working to finish it soon.

I'll check what the eta is.

[astubbs]

ok @JorgenRingen , issue has been raised internally - and we should have a public PR builder setup soon, along with matrix builds for multiple AK versions like we used to have with GH actions.

[astubbs]

FYI Matrix test against multiple AK and JDK versions in GH #103

[astubbs]

Ok @JorgenRingen , if you rebase now it should build