gharris1727 commented on code in PR #13178:
URL: https://github.com/apache/kafka/pull/13178#discussion_r1107628693
##########
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java:
##########
@@ -701,43 +745,151 @@ protected void produceMessages(EmbeddedConnectCluster
cluster, String topicName,
int cnt = 0;
for (int r = 0; r < NUM_RECORDS_PER_PARTITION; r++)
for (int p = 0; p < numPartitions; p++)
- cluster.kafka().produce(topicName, p, "key", "value-" + cnt++);
+ produce(cluster.kafka(), topicName, p, "key", "value-" +
cnt++);
}
-
+
+ /**
+ * Produce a test record to a Kafka cluster.
+ * This method allows subclasses to configure and use their own Kafka
Producer instead of using the built-in default.
+ * @param cluster Kafka cluster that should receive the record
+ * @param topic Topic to send the record to, non-null
+ * @param partition Partition to send the record to, maybe null.
+ * @param key Kafka key for the record
+ * @param value Kafka value for the record
+ */
+ protected void produce(EmbeddedKafkaCluster cluster, String topic, Integer
partition, String key, String value) {
+ cluster.produce(topic, partition, key, value);
+ }
+
+ protected static Map<TopicPartition, OffsetAndMetadata>
waitForCheckpointOnAllPartitions(
+ MirrorClient client, String consumerGroupName, String
remoteClusterAlias, String topicName
+ ) throws InterruptedException {
+ AtomicReference<Map<TopicPartition, OffsetAndMetadata>> ret = new
AtomicReference<>();
+ waitForCondition(
+ () -> {
+ Map<TopicPartition, OffsetAndMetadata> offsets =
client.remoteConsumerOffsets(
+ consumerGroupName, remoteClusterAlias,
Duration.ofMillis(3000));
+ for (int i = 0; i < NUM_PARTITIONS; i++) {
+ if (!offsets.containsKey(new TopicPartition(topicName,
i))) {
+ log.info("Checkpoint is missing for {}: {}-{}",
consumerGroupName, topicName, i);
+ return false;
+ }
+ }
+ ret.set(offsets);
+ return true;
+ },
+ CHECKPOINT_DURATION_MS,
+ String.format(
+ "Offsets for consumer group %s not translated from %s
for topic %s",
+ consumerGroupName,
+ remoteClusterAlias,
+ topicName
+ )
+ );
+ return ret.get();
+ }
+
/*
* given consumer group, topics and expected number of records, make sure
the consumer group
* offsets are eventually synced to the expected offset numbers
*/
- protected static <T> void
waitForConsumerGroupOffsetSync(EmbeddedConnectCluster connect,
- Consumer<T, T> consumer, List<String> topics, String
consumerGroupId, int numRecords, boolean exactOffsetTranslation)
- throws InterruptedException {
+ protected static <T> void waitForConsumerGroupFullSync(
+ EmbeddedConnectCluster connect, List<String> topics, String
consumerGroupId, int numRecords, boolean exactOffsetTranslation
+ ) throws InterruptedException {
try (Admin adminClient = connect.kafka().createAdminClient()) {
- List<TopicPartition> tps = new ArrayList<>(NUM_PARTITIONS *
topics.size());
+ Map<TopicPartition, OffsetSpec> tps = new HashMap<>(NUM_PARTITIONS
* topics.size());
for (int partitionIndex = 0; partitionIndex < NUM_PARTITIONS;
partitionIndex++) {
for (String topic : topics) {
- tps.add(new TopicPartition(topic, partitionIndex));
+ tps.put(new TopicPartition(topic, partitionIndex),
OffsetSpec.latest());
}
}
long expectedTotalOffsets = numRecords * topics.size();
waitForCondition(() -> {
Map<TopicPartition, OffsetAndMetadata> consumerGroupOffsets =
adminClient.listConsumerGroupOffsets(consumerGroupId).partitionsToOffsetAndMetadata().get();
- long consumerGroupOffsetTotal =
consumerGroupOffsets.values().stream()
+ long totalConsumerGroupOffsets =
consumerGroupOffsets.values().stream()
.mapToLong(OffsetAndMetadata::offset).sum();
- Map<TopicPartition, Long> offsets = consumer.endOffsets(tps,
CONSUMER_POLL_TIMEOUT_MS);
- long totalOffsets = offsets.values().stream().mapToLong(l ->
l).sum();
-
+ Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo>
endOffsets =
+ adminClient.listOffsets(tps).all().get();
+ long totalEndOffsets = endOffsets.values().stream()
+
.mapToLong(ListOffsetsResult.ListOffsetsResultInfo::offset).sum();
+
+ for (TopicPartition tp : endOffsets.keySet()) {
+ if (consumerGroupOffsets.containsKey(tp)) {
+ assertTrue(consumerGroupOffsets.get(tp).offset() <=
endOffsets.get(tp).offset(),
+ "Consumer group committed downstream offsets
beyond the log end, this would lead to negative lag metrics"
+ );
+ }
+ }
boolean totalOffsetsMatch = exactOffsetTranslation
- ? totalOffsets == expectedTotalOffsets
- : totalOffsets >= expectedTotalOffsets;
+ ? totalEndOffsets == expectedTotalOffsets
+ : totalEndOffsets >= expectedTotalOffsets;
+
+ boolean consumerGroupOffsetsMatch = exactOffsetTranslation
+ ? totalConsumerGroupOffsets == expectedTotalOffsets
+ : totalConsumerGroupOffsets >= expectedTotalOffsets;
// make sure the consumer group offsets are synced to expected
number
- return totalOffsetsMatch && consumerGroupOffsetTotal > 0;
+ return totalOffsetsMatch && consumerGroupOffsetsMatch;
}, OFFSET_SYNC_DURATION_MS, "Consumer group offset sync is not
complete in time");
}
}
+ protected static void insertDummyOffsetSyncs(EmbeddedConnectCluster
cluster, String offsetSyncsTopic, String topic, int partitions) throws
InterruptedException {
+ waitForTopicCreated(cluster, offsetSyncsTopic);
+ // Insert a large number of checkpoint records into the offset syncs
topic to simulate
+ // a long-lived MM2 instance that has replicated many offsets in the
past.
+ List<ProducerRecord<byte[], byte[]>> records = new ArrayList<>();
+ for (int partition = 0; partition < partitions; partition++) {
+ TopicPartition tp = new TopicPartition(topic, partition);
+ OffsetSync sync = new OffsetSync(tp, 0L, 0L);
+ records.add(new ProducerRecord<>(offsetSyncsTopic,
sync.recordKey(), sync.recordValue()));
+ }
+ Map<String, Object> producerProps = new HashMap<>();
+ int sentRecords = 0;
+ try (KafkaProducer<byte[], byte[]> producer =
cluster.kafka().createProducer(producerProps)) {
+ // Try to ensure that the contents of the offset syncs topic is
too large to read before the checkpoint
+ // interval passes, so that the first checkpoint would take place
before reading the whole contents of the
+ // sync topic. Experimentally, this test passes with <2x, and
fails with >5x, without a fix for KAFKA-13659.
+ double consumeEfficiency = 10;
+ long deadline = System.currentTimeMillis() + (long)
(consumeEfficiency * CHECKPOINT_INTERVAL_DURATION_MS);
+ int nRecords = records.size();
+ while (System.currentTimeMillis() < deadline) {
+ producer.send(records.get(sentRecords % nRecords));
+ sentRecords++;
+ }
+ producer.flush();
+ }
+ log.info("Sent {} dummy records to {}", sentRecords, offsetSyncsTopic);
+ }
Review Comment:
I'll remove insertDummyOffsetSyncs, but leave the testRestartReplication, as
it is still testing a new situation: restarting the connectors mid-replication.
It won't target this specific bug without the `insertDummyOffsetSyncs`, but
could still add value.
I've added unit tests for OffsetSyncStore that address the bug (translating
before start finishes) that should hopefully compensate for this test's smaller
scope.
I looked into testing MirrorCheckpointTask, and it looks too involved to
address in this PR. I think it would be more appropriate to follow up later
with some refactoring & testing for the checkpoint task.
I feel confident from the results of the insertDummyOffsetSyncs test that:
1. We are addressing the situation where the offset syncs topic is very large
2. We are not introducing a severe performance regression, as the
consumeEfficiency is very high.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
