[GitHub] [kafka] vvcephei commented on a change in pull request #9368: KAFKA-9274: Add timeout handling for state restore

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

vvcephei commented on a change in pull request #9368:
URL: https://github.com/apache/kafka/pull/9368#discussion_r499812565



##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
##########
@@ -205,4 +207,9 @@ default boolean maybePunctuateSystemTime() {
         return false;
     }
 
+    void maybeInitTaskTimeoutOrThrow(final long currentWallClockMs,
+                                     final TimeoutException timeoutException,
+                                     final Logger log) throws StreamsException;

Review comment:
       StreamsException is unchecked, right?
   
   It's better to document unchecked exceptions in the `@throws` javadoc tag. The `throws` keyword is for telling the compiler that you want callers instead of yourself to handle a _checked_ exception. I honestly have no idea why the java team chose to say "this is poor style" instead of just making it a compiler error, but that's the rationale. https://www.oracle.com/technical-resources/articles/java/javadoc-tool.html#throwstag

##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
##########
@@ -431,6 +432,9 @@ public void restore() {
                 // in order to make sure we call the main consumer#poll in time.
                 // TODO: once we move ChangelogReader to a separate thread this may no longer be a concern
                 polledRecords = restoreConsumer.poll(state == ChangelogReaderState.STANDBY_UPDATING ? Duration.ZERO : pollTime);
+
+                // TODO (?) If we cannot fetch records during restore, should we trigger `task.timeout.ms` ?
+                // TODO (?) If we cannot fetch records for standby task, should we trigger `task.timeout.ms` ?

Review comment:
       Thanks. I agree we can afford to leave this for future improvements.
   
   It does seem like we should have some kind of improvement in the future, though. Having a restore or standby-update fail indefinitely would be just as damaging to an application's robustness as having the main consumer fail indefinitely.
   
   Perhaps we can make some improvements to the Consumer first, though, so that we don't have to do so much guesswork to distinguish between "no records" and "no fetch".

##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
##########
@@ -137,4 +146,48 @@ public void update(final Set<TopicPartition> topicPartitions, final Map<String,
         this.inputPartitions = topicPartitions;
         topology.updateSourceTopics(nodeToSourceTopics);
     }
+
+    @Override
+    public void maybeInitTaskTimeoutOrThrow(final long currentWallClockMs,
+                                            final TimeoutException timeoutException,
+                                            final Logger log) throws StreamsException {
+        if (deadlineMs == NO_DEADLINE) {
+            deadlineMs = currentWallClockMs + taskTimeoutMs;
+        } else if (currentWallClockMs > deadlineMs) {
+            final String errorMessage = String.format(
+                "Task %s did not make progress within %d ms. Adjust `%s` if needed.",
+                id,
+                currentWallClockMs - deadlineMs + taskTimeoutMs,
+                StreamsConfig.TASK_TIMEOUT_MS_CONFIG
+            );
+
+            if (timeoutException != null) {
+                throw new TimeoutException(errorMessage, timeoutException);
+            } else {
+                throw new TimeoutException(errorMessage);
+            }
+        }
+
+        if (timeoutException != null) {
+            log.debug(
+                "Timeout exception. Remaining time to deadline {}; retrying.",
+                deadlineMs - currentWallClockMs,
+                timeoutException
+            );

Review comment:
       This is the wrong format for this log message. The exception won't be logged. You have to format the string first:
   ```suggestion
               log.debug(
                   String.format("Timeout exception. Remaining time to deadline %d; retrying.",
                   deadlineMs - currentWallClockMs),
                   timeoutException
               );
   ```

##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
##########
@@ -137,4 +146,48 @@ public void update(final Set<TopicPartition> topicPartitions, final Map<String,
         this.inputPartitions = topicPartitions;
         topology.updateSourceTopics(nodeToSourceTopics);
     }
+
+    @Override
+    public void maybeInitTaskTimeoutOrThrow(final long currentWallClockMs,
+                                            final TimeoutException timeoutException,
+                                            final Logger log) throws StreamsException {

Review comment:
       Sounds good. An alternative is to add an `abstract Logger log()` to AbstractTask's interface, which would make it clearer that the logger is still going to have the appropriate class name.

##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
##########
@@ -694,7 +694,7 @@
                     CommonClientConfigs.SECURITY_PROTOCOL_DOC)
             .define(TASK_TIMEOUT_MS_CONFIG,
                     Type.LONG,
-                    Duration.ofSeconds(5L).toMillis(),
+                    Duration.ofMinutes(5L).toMillis(),

Review comment:
       Oops! Good catch.




----------------------------------------------------------------
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.

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