KAFKA-9645: Fallback to unsubscribe during Task Migrated

[abbccdda]

After #7312, we could still return data during the rebalance phase, which means it could be possible to find records without corresponding tasks. We have to fallback to the unsubscribe mode during task migrated as the assignment should be cleared out to keep sync with task manager state.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[guozhangwang]

The issue is that when handling TaskMigrated, we call handleLostAll which closes all tasks, and then call enforceRebalance which sets the flag. However inside consumer the assignment are not cleared yet and would still contain the assigned partitions. That why in the next poll call, we could still return the data for those still-assigned partitions --- inside consumers there's a filtering that we would not return data that are from no longer assigned partitions --- and then hit this issue.

What I'm thinking now is, should we reset the assignment inside consumer when enforceRebalance, which would cause the subscription's owned partitions to be empty. If it is due to task-migrated, then it is fine since we know we have likely lost those partitions anyways; if it is only due to version probing then we still want to encode those owned partitions. Thoughts? cc @ableegoldman

[abbccdda]

Thanks Guozhang for the summary, I think there is also an expectation around the enforceRebalance API, which is whether we should set the AbstractCoordinator state to REBALANCING immediately as if we don't see a coordinator, we shall retry discovering but during this period, the state of consumer is not correct. This could be an orthogonal fix to the consumer in the meantime if you think this makes sense @ableegoldman

[abbccdda]

Double checked with unit test that under old code path, we shall hit the reported exception:

java.lang.NullPointerException: Task was unexpectedly missing for partition topic1-1
	at org.apache.kafka.streams.processor.internals.StreamThread.addRecordsToTasks(StreamThread.java:986)
	at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:822)
	at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:751)
	at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:725)
	at org.junit.Assert.assertThrows(Assert.java:1001)
	at org.junit.Assert.assertThrows(Assert.java:981)

[ableegoldman]

Thanks Boyang, this LGTM

[guozhangwang]

Test this please

[guozhangwang]

test this please

[guozhangwang]

test this please

[guozhangwang]

test this please

[guozhangwang]

test this please

[guozhangwang]

LGTM! After green build I will merge to trunk.

[guozhangwang]

Nice.

[guozhangwang]

Hmm :) it reminds me that the mock consumer's behavior is not exactly the same as the actual consumer (the later would filter, the former would throw), but perhaps this worth a different PR to cleanup. @abbccdda could you file a JIRA for it?

[abbccdda]

Sounds good to me! https://issues.apache.org/jira/browse/KAFKA-9679

[guozhangwang]

test this please