KAFKA-15607:Possible NPE is thrown in MirrorCheckpointTask

[hudeqi]

In the syncGroupOffset method, if targetConsumerOffset.get(topicPartition) gets null, then the calculation of latestDownstreamOffset will throw NPE. This usually occurs in this situation: a group consumed a topic in the target cluster previously. Later, the group offset of some partitions was reset to -1, the OffsetAndMetadata of these partitions was null.

It is possible that when reset offsets are performed in the java kafka client, the reset to -1 will be intercepted. However, there are some other types of clients such as sarama, which can magically reset the group offset to -1, so MM2 will trigger an NPE exception in this scenario. Therefore, a defensive measure to avoid NPE is needed here.

[hudeqi]

Hi, @clolov. Excuse me, can you help me trigger the build for this PR? I'm also collaborator, why can't I find a place to trigger build PR :)

[C0urante]

Thanks @hudeqi, I think this is reasonable. Do you know why Sarama sets offsets to -1? If it's for normal operations and not indicative of something wrong, we may not even need to log a warning message in that case and could change the check here from !targetConsumerOffset.containsKey(topicPartition) to targetConsumerOffset.get(topicPartition) == null.

[hudeqi]

Thanks @hudeqi, I think this is reasonable. Do you know why Sarama sets offsets to -1? If it's for normal operations and not indicative of something wrong, we may not even need to log a warning message in that case and could change the check here from !targetConsumerOffset.containsKey(topicPartition) to targetConsumerOffset.get(topicPartition) == null.

Thanks your review. @C0urante . In fact, directly resetting to -1 is an abnormal operation, whether for Sarama or other clients. This problem was discovered in this way: when using the Sarama client, we wanted to reset the group's offset to the latest, so we passed in the OffsetNewest in Sarama as a parameter to call the reset offset method. Finally, it was discovered that the offset was reset to -1. The reason is that the value of OffsetNewest is -1. For Sarama, resetting to the latest should be another operation process, but this kind of misoperation is not intercepted like the java client, which can be deal friendly. So this issue occurred when encountering scenarios like MM2. So I think it is better to add a warn log here.

[C0urante]

Makes sense, thanks for the explanation @hudeqi!