KAFKA-9610: Do not throw illegal state when remaining partitions are not empty

[abbccdda]

For handleRevocation, it is possible that previous onAssignment callback has cleaned up the stream tasks, which means no corresponding task could be found for given partitions. We should not throw here as this is expected behavior.

Committer Checklist (excluded from commit message)

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

[ableegoldman]

This doesn't really make sense, onPartitionsRevoked is always called before onAssignment

[ableegoldman]

Not sure if we want to log as warn or debug though -- hitting this indicates we hit the (likely not that common) race condition, or a bug.

[abbccdda]

@guozhangwang could you also take a look and see if you could trigger a test?

[guozhangwang]

test this please

[guozhangwang]

LGTM.

[abbccdda]

The failure was really weird as it is not related to the change:

20:20:40 > Task :streams:compileTestJava
20:20:40 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java:351: error: method getAllTopicsInCluster in class KafkaZkClient cannot be applied to given types;
20:20:40         return JavaConverters.setAsJavaSetConverter(brokers[0].kafkaServer().zkClient().getAllTopicsInCluster()).asJava();
20:20:40                                                                                        ^
20:20:40   required: boolean
20:20:40   found: no arguments
20:20:40   reason: actual and formal argument lists differ in length
20:20:49 1 error

Will rebase trunk and see if this fixes

[abbccdda]

https://builds.apache.org/job/kafka-pr-jdk11-scala2.13/4902/console

[omkreddy]

The failure was really weird as it is not related to the change:

20:20:40 > Task :streams:compileTestJava
20:20:40 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.13/streams/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java:351: error: method getAllTopicsInCluster in class KafkaZkClient cannot be applied to given types;
20:20:40         return JavaConverters.setAsJavaSetConverter(brokers[0].kafkaServer().zkClient().getAllTopicsInCluster()).asJava();
20:20:40                                                                                        ^
20:20:40   required: boolean
20:20:40   found: no arguments
20:20:40   reason: actual and formal argument lists differ in length
20:20:49 1 error

Will rebase trunk and see if this fixes

Fixed in #8170

[guozhangwang]

test this please

[guozhangwang]

test this please