fix concurrency issue in application shutdown
Need to ensure that `enforceRebalance` is used in a thread safe way
Reviewers: Bruno Cadonna Anna Sophie Blee-Goldman
@@ -571,6 +571,11 @@ public class StreamThread extends Thread {
// 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();
+ }
runOnce();
if (nextProbingRebalanceMs.get() < time.milliseconds()) {
log.info("Triggering the followup rebalance scheduled for {} ms.", nextProbingRebalanceMs.get());
@@ -660,10 +665,7 @@ public class StreamThread extends Thread {
}
public void sendShutdownRequest(final AssignorError assignorError) {
- log.warn("Detected that shutdown was requested. " +
- "All clients in this app will now begin to shutdown");
assignmentErrorCode.set(assignorError.code());
- mainConsumer.enforceRebalance();
}
private void handleTaskMigrated(final TaskMigratedException e) {
