[GitHub] [kafka] wcarlson5 commented on a change in pull request #11381: KAFKA-12648: allow users to set a StreamsUncaughtExceptionHandler on individual named topologies

[GitBox <gi...@apache.org>]

wcarlson5 commented on a change in pull request #11381:
URL: https://github.com/apache/kafka/pull/11381#discussion_r722408690



##########
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:
       Do we want to swallow if it is not a named topology? should we not just rethrow the error in that case?

##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -299,6 +301,7 @@ public boolean isRunning() {
     private final java.util.function.Consumer<Long> cacheResizer;
 
     private java.util.function.Consumer<Throwable> streamsUncaughtExceptionHandler;
+    private final Map<String, java.util.function.Consumer<Throwable>> topologyExceptionHandlers;

Review comment:
       Do we want to be able to clean these up after we removing a topology? We might not but it might also be a good idea.
   
   Example we remove a topology with a handler and replace one with the same name without a handler or using the non named topology handler




-- 
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: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org