mjsax commented on a change in pull request #10417:
URL: https://github.com/apache/kafka/pull/10417#discussion_r602590947
##########
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 (processingMode == ProcessingMode.EXACTLY_ONCE_ALPHA ||
processingMode == ProcessingMode.EXACTLY_ONCE_BETA) {
Review comment:
Why are we removing this condition?
--
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:
us...@infra.apache.org
