ableegoldman commented on a change in pull request #10417:
URL: https://github.com/apache/kafka/pull/10417#discussion_r602595373
##
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##
@@ -565,59 +565,57 @@ public void run() {
* @throws StreamsException if the store's change log does not
contain the partition
*/
boolean runLoop() {
-subscribeConsumer();
+try {
+subscribeConsumer();
-// if the thread is still in the middle of a rebalance, we should keep
polling
-// until the rebalance is completed before we close and commit the
tasks
-while (isRunning() || taskManager.isRebalanceInProgress()) {
-try {
-if (assignmentErrorCode.get() ==
AssignorError.SHUTDOWN_REQUESTED.code()) {
-log.warn("Detected that shutdown was requested. " +
-"All clients in this app will now begin to
shutdown");
-mainConsumer.enforceRebalance();
-}
-final Long size = cacheResizeSize.getAndSet(-1L);
-if (size != -1L) {
-cacheResizer.accept(size);
-}
-runOnce();
-if (nextProbingRebalanceMs.get() < time.milliseconds()) {
-log.info("Triggering the followup rebalance scheduled for
{} ms.", nextProbingRebalanceMs.get());
-mainConsumer.enforceRebalance();
-nextProbingRebalanceMs.set(Long.MAX_VALUE);
-}
-} catch (final TaskCorruptedException e) {
-log.warn("Detected the states of tasks " + e.corruptedTasks()
+ " are corrupted. " +
-"Will close the task as dirty and re-create and
bootstrap from scratch.", e);
+// if the thread is still in the middle of a rebalance, we should
keep polling
+// until the rebalance is completed before we close and commit the
tasks
+while (isRunning() || taskManager.isRebalanceInProgress()) {
try {
-taskManager.handleCorruption(e.corruptedTasks());
-} catch (final TaskMigratedException taskMigrated) {
-handleTaskMigrated(taskMigrated);
-}
-} catch (final TaskMigratedException e) {
-handleTaskMigrated(e);
-} catch (final UnsupportedVersionException e) {
-final String errorMessage = e.getMessage();
-if (errorMessage != null &&
+if (assignmentErrorCode.get() ==
AssignorError.SHUTDOWN_REQUESTED.code()) {
+log.warn("Detected that shutdown was requested. " +
+ "All clients in this app will now begin
to shutdown");
+mainConsumer.enforceRebalance();
+}
+final Long size = cacheResizeSize.getAndSet(-1L);
+if (size != -1L) {
+cacheResizer.accept(size);
+}
+runOnce();
+if (nextProbingRebalanceMs.get() < time.milliseconds()) {
+log.info("Triggering the followup rebalance scheduled
for {} ms.", nextProbingRebalanceMs.get());
+mainConsumer.enforceRebalance();
+nextProbingRebalanceMs.set(Long.MAX_VALUE);
+}
+} catch (final TaskCorruptedException e) {
+log.warn("Detected the states of tasks " +
e.corruptedTasks() + " are corrupted. " +
+ "Will close the task as dirty and re-create
and bootstrap from scratch.", e);
+try {
+taskManager.handleCorruption(e.corruptedTasks());
+} catch (final TaskMigratedException taskMigrated) {
+handleTaskMigrated(taskMigrated);
+}
+} catch (final TaskMigratedException e) {
+handleTaskMigrated(e);
+} catch (final UnsupportedVersionException e) {
+final String errorMessage = e.getMessage();
+if (errorMessage != null &&
errorMessage.startsWith("Broker unexpectedly doesn't
support requireStable flag on version ")) {
-log.error("Shutting down because the Kafka cluster seems
to be on a too old version. " +
-"Setting {}=\"{}\" requires broker version
2.5 or higher.",
-StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
-EXACTLY_ONCE_BETA);
-}
-failedStreamThreadSensor.record();
-this.streamsUncaughtExceptionHandler.accept(e);
-if (proces
