lianetm commented on code in PR #22035:
URL: https://github.com/apache/kafka/pull/22035#discussion_r3453217786
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java:
##########
@@ -155,4 +159,32 @@ public void close(final Duration timeout) {
() -> log.warn("The application event handler was already
closed")
);
}
+
+ /**
+ * Best-effort check that the consumer network thread is still alive. If
the thread has
+ * already terminated (due to a failure or shutdown), it will never
process any events from
+ * the queue. Rather than blocking indefinitely or timing out with a
misleading error, this
+ * fails fast with a clear error message.
+ *
+ * <p>Note: this is inherently racy — the thread could die between this
check and the
+ * subsequent {@code applicationEventQueue.add()}. That narrow window is
acceptable because
+ * any subsequent call to {@code add()} will detect the dead thread
immediately, and any
+ * orphaned events will be expired by the {@link CompletableEventReaper}
during consumer
+ * {@link #close() close}.
+ *
+ * @throws KafkaException if the background thread is not alive
+ */
+ private void ensureNetworkThreadAlive() {
+ if (networkThread == null || !networkThread.isAlive()) {
+ String message = "The consumer background thread is not running
and cannot process requests.";
+ Optional<Throwable> terminationError = networkThread != null
+ ? networkThread.terminationError()
+ : Optional.empty();
+ if (terminationError.isPresent()) {
+ Throwable cause = terminationError.get();
+ throw new KafkaException(message + " Cause: " +
cause.getMessage(), cause);
+ }
Review Comment:
related to above msg, can we simplify here and remove this? (just check if
networkThread.isAlive)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -166,6 +167,7 @@ public void run() {
}
}
} catch (Throwable t) {
+ maybeSetFatalError(t);
Review Comment:
are we ever going to reach this to set this fatalError? seems not, this
catch covers 2 actions: initialization (has its own catch and returns), and the
runOnce (also has it's own catch and swallows). This makes me wonder if we
really need this fatal error, or maybe we just need to have the
ensureNetworkThreadAlive check if thread.isAlive. Wdyt?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -179,6 +181,31 @@ private void maybeSetInitializationError(KafkaException
error) {
log.error("Consumer network thread resource initialization error ({})
will be suppressed as an error was already set", error.getMessage(), error);
}
+ private void maybeSetFatalError(Throwable error) {
+ if (fatalError.compareAndSet(null, error))
+ return;
+
+ log.error("Consumer network thread fatal error ({}) will be suppressed
as an error was already set", error.getMessage(), error);
+ }
+
+ /**
+ * Returns the error that caused the network thread to terminate, if any.
This may be a runtime
+ * failure ({@link #fatalError}) or an initialization failure ({@link
#initializationError}).
+ * The thread object remains accessible after the thread has exited, so
the application thread
+ * can retrieve this when subsequent API calls detect the thread is no
longer alive.
+ */
+ public Optional<Throwable> terminationError() {
Review Comment:
related to above, can all this be removed too?
--
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]
