ableegoldman commented on a change in pull request #11381:
URL: https://github.com/apache/kafka/pull/11381#discussion_r722488122
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
##########
@@ -1280,29 +1281,36 @@ int process(final int maxNumRecords, final Time time) {
int processed = 0;
final long then = now;
try {
- while (processed < maxNumRecords && task.process(now)) {
- task.clearTaskTimeout();
- processed++;
+ try {
+ while (processed < maxNumRecords && task.process(now)) {
+ task.clearTaskTimeout();
+ processed++;
+ }
+ } catch (final TimeoutException timeoutException) {
+ task.maybeInitTaskTimeoutOrThrow(now, timeoutException);
+ log.debug(
+ String.format(
+ "Could not complete processing records for %s due
to the following exception; will move to next task and retry later",
+ task.id()),
+ timeoutException
+ );
+ } catch (final TaskMigratedException e) {
+ log.info("Failed to process stream task {} since it got
migrated to another thread already. " +
+ "Will trigger a new rebalance and close all
tasks as zombies together.", task.id());
+ throw e;
+ } catch (final RuntimeException e) {
+ log.error("Failed to process stream task {} due to the
following error:", task.id(), e);
+ throw e;
+ } finally {
+ now = time.milliseconds();
+ totalProcessed += processed;
+ task.recordProcessBatchTime(now - then);
+ }
+ } catch (final Throwable e) {
+ final String topologyName = task.id().topologyName();
Review comment:
Fixed 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
