root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 5 Packages Projects Releases Wiki Activity

KAFKA-15607: Fix NPE in MirrorCheckpointTask::syncGroupOffset (#14587) 

Reviewers: Chris Egerton <chrise@aiven.io>
This commit is contained in:
hudeqi 2023-10-21 00:17:51 +08:00 committed by GitHub
parent b5ec6e8a0d
commit 4083cd627e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 41 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorCheckpointTask.java
View File

@ -330,11 +330,20 @@ public class MirrorCheckpointTask extends SourceTask {
// if translated offset from upstream is smaller than the current consumer offset
// in the target, skip updating the offset for that partition
long latestDownstreamOffset = targetConsumerOffset.get(topicPartition).offset();
if (latestDownstreamOffset >= convertedOffset.offset()) {
log.trace("latestDownstreamOffset {} is larger than or equal to convertedUpstreamOffset {} for "
+ "TopicPartition {}", latestDownstreamOffset, convertedOffset.offset(), topicPartition);
continue;
OffsetAndMetadata targetOffsetAndMetadata = targetConsumerOffset.get(topicPartition);
if (targetOffsetAndMetadata != null) {
long latestDownstreamOffset = targetOffsetAndMetadata.offset();
if (latestDownstreamOffset >= convertedOffset.offset()) {
log.trace("latestDownstreamOffset {} is larger than or equal to convertedUpstreamOffset {} for "
+ "TopicPartition {}", latestDownstreamOffset, convertedOffset.offset(), topicPartition);
continue;
}
} else {
// It is possible that when resetting 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, which will cause
// `targetOffsetAndMetadata` here is null. For this case, just sync the offset to target.
log.warn("Group {} offset for partition {} may has been reset to a negative offset, just sync the offset to target.",
consumerGroupId, topicPartition);
}
offsetToSync.put(topicPartition, convertedOffset);
}

27
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorCheckpointTaskTest.java
View File

@ -169,6 +169,33 @@ public class MirrorCheckpointTaskTest {
"Consumer 2 " + topic2 + " failed");
}
@Test
public void testSyncOffsetForTargetGroupWithNullOffsetAndMetadata() {
Map<String, Map<TopicPartition, OffsetAndMetadata>> idleConsumerGroupsOffset = new HashMap<>();
Map<String, Map<TopicPartition, Checkpoint>> checkpointsPerConsumerGroup = new HashMap<>();
String consumer = "consumer";
String topic = "topic";
Map<TopicPartition, OffsetAndMetadata> ct = new HashMap<>();
TopicPartition tp = new TopicPartition(topic, 0);
// Simulate other clients such as Sarama, which may reset group offsets to -1. This can cause
// the obtained `OffsetAndMetadata` of the target cluster to be null.
ct.put(tp, null);
idleConsumerGroupsOffset.put(consumer, ct);
Checkpoint cp = new Checkpoint(consumer, new TopicPartition(topic, 0), 200, 101, "metadata");
Map<TopicPartition, Checkpoint> checkpointMap = new HashMap<>();
checkpointMap.put(cp.topicPartition(), cp);
checkpointsPerConsumerGroup.put(consumer, checkpointMap);
MirrorCheckpointTask mirrorCheckpointTask = new MirrorCheckpointTask("source", "target",
new DefaultReplicationPolicy(), null, idleConsumerGroupsOffset, checkpointsPerConsumerGroup);
Map<String, Map<TopicPartition, OffsetAndMetadata>> output = mirrorCheckpointTask.syncGroupOffset();
assertEquals(101, output.get(consumer).get(tp).offset(), "Consumer " + topic + " failed");
}
@Test
public void testNoCheckpointForTopicWithoutOffsetSyncs() {
OffsetSyncStoreTest.FakeOffsetSyncStore offsetSyncStore = new OffsetSyncStoreTest.FakeOffsetSyncStore();