Need to flush state even on unclean close
In the case of unclean close we still need to make sure all the stores are flushed before closing any.
Reviewers: Matthias J. Sax Boyang Chen Bill Bejeck Guozhang Wang
@@ -678,12 +678,17 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator
throw taskMigratedException;
}
} else {
+ // In the case of unclean close we still need to make sure all the stores are flushed before closing any
+ super.flushState();
+
+ if (eosEnabled) {
maybeAbortTransactionAndCloseRecordCollector(isZombie);
}
}
+ }
private void maybeAbortTransactionAndCloseRecordCollector(final boolean isZombie) {
- if (eosEnabled && !isZombie) {
+ if (!isZombie) {
try {
if (transactionInFlight) {
producer.abortTransaction();
@@ -701,7 +706,6 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator
}
}
- if (eosEnabled) {
try {
recordCollector.close();
} catch (final Throwable e) {
@@ -710,7 +714,6 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator
producer = null;
}
}
- }
private void closeTopology() {
log.trace("Closing processor topology");
