MINOR: log and fail on missing task in Streams
Matthias J. Sax Guozhang Wang
@@ -979,7 +979,14 @@ public class StreamThread extends Thread {
for (final TopicPartition partition : records.partitions()) {
final StreamTask task = taskManager.activeTask(partition);
- if (task.isClosed()) {
+ if (task == null) {
+ log.error(
+ "Unable to locate active task for received-record partition {}. Current tasks: {}",
+ partition,
+ taskManager.toString(">")
+ );
+ throw new NullPointerException("Task was unexpectedly missing for partition " + partition);
+ } else if (task.isClosed()) {
log.info("Stream task {} is already closed, probably because it got unexpectedly migrated to another thread already. " +
"Notifying the thread to trigger a new rebalance immediately.", task.id());
throw new TaskMigratedException(task);
