[
https://issues.apache.org/jira/browse/KAFKA-19107?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18101182#comment-18101182
]
Yunseop Eom commented on KAFKA-19107:
-------------------------------------
Opened PR #23054: https://github.com/apache/kafka/pull/23054
The change handles alterConsumerGroupOffsets results per partition. When a
checkpoint targets a deleted topic-partition, MirrorCheckpointTask stops
retrying it, emits a checkpoint tombstone, and CheckpointStore applies that
tombstone when loading the compacted checkpoints topic. Other partitions in the
same request continue to sync normally, and checkpoint-store access is
synchronized for the poll/scheduler concurrency.
Validation completed:
- RED regression tests added before implementation.
- MirrorCheckpointTaskTest and CheckpointStoreTest pass, including no-repeat,
mixed-partition, tombstone emission, and restart cleanup cases.
- mirror-client and mirror module Spotless, Checkstyle, SpotBugs, and git diff
checks pass.
The PR is ready for review.
> MirrorMaker2 repeatedly tries to sync group offsets for partitions that no
> longer exist
> ---------------------------------------------------------------------------------------
>
> Key: KAFKA-19107
> URL: https://issues.apache.org/jira/browse/KAFKA-19107
> Project: Kafka
> Issue Type: Bug
> Components: mirrormaker
> Affects Versions: 3.9.0
> Reporter: Alex Rovner
> Priority: Major
>
> h3. Summary
> MM2's Checkpoint connector is currently unaware of topics being deleted and
> will repeatedly attempt to sync group offsets for them, which results in
> `UnknownTopicOrPartitionException` being logged at each sync. The checkpoints
> used for this sync are persisted in a MM2-internal checkpoints topic. Because
> the topic is compacted, the checkpoints that reference deleted
> topic-partitions are never cleaned up. This means that
> `UnknownTopicOrPartitionException` will occur indefinitely on every sync
> until manual intervention.
> Our current workaround is to change the retention policy of the checkpoints
> topic to `compact,delete` or even `delete` to make sure that old checkpoints
> do not remain in the topic forever. However, this is not an ideal approach
> because (correct me if I'm wrong) any cleanup would only take effect when the
> connector is restarted because that is when the contents of the
> checkpoint-topic are loaded into the internal state of the connector.
> h3. How to reproduce
> # Create 2 Kafka clusters (my-cluster and my-backup) with
> `{{{}auto.create.topics.enable{}}}=false`
> # Create a MM2 instance and configure it to sync consumer group offsets
> # Spin up a dummy synthetic client that continuously produces to some topic
> (e.g. `synth-client-topic`)
> # Consume from `synth-client-topic` using `kafka-console-consumer` with
> `--group foo`
> # Consume from the checkpoints topic to make sure that some checkpoints have
> been created:
> {{kafka-console-consumer.sh --bootstrap-server my-backup-kafka-bootstrap:9092
> --topic source.checkpoints.internal --formatter
> "org.apache.kafka.connect.mirror.formatters.CheckpointFormatter"
> --from-beginning}}
> # Shut down MM2, consume some more from the `synth-client-topic`. This is
> probably not strictly necessary, but we want to create a scenario where the
> groups in the two clusters are out-of-sync.
> # Stop the dummy client. Delete the source and mirrored topics.
> # Launch MM2 again. Observe that it is now logging errors like this one
> every minute:
> {quote}{{{}2025-04-07 12:25:34,203 ERROR
> [source->backup.MirrorCheckpointConnector|task-0] Unable to sync offsets for
> consumer group foo. (org.apache.kafka.connect.mirror.MirrorCheckpointTask)
> [kafka-admin-client-thread |
> source->backup|source->backup.MirrorCheckpointConnector-0|checkpoint-target-admin]{}}}{{{}java.util.concurrent.CompletionException:
> org.apache.kafka.common.errors.UnknownTopicOrPartitionException: Failed
> altering consumer group offsets for the following partitions:
> [source.synth-client-topic-0]{}}}{{ at
> java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:315)}}{{
> at
> java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:320)}}{{
> at
> java.base/java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:649)}}{{
> at
> java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510)}}{{
> at
> java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147)}}{{
> at
> org.apache.kafka.common.internals.KafkaCompletableFuture.kafkaComplete(KafkaCompletableFuture.java:39)}}{{
> at
> org.apache.kafka.common.internals.KafkaFutureImpl.complete(KafkaFutureImpl.java:122)}}{{
> at
> org.apache.kafka.clients.admin.internals.AdminApiFuture$SimpleAdminApiFuture.complete(AdminApiFuture.java:100)}}{{
> at
> java.base/java.util.Collections$SingletonMap.forEach(Collections.java:5061)}}{{
> at
> java.base/java.util.Collections$UnmodifiableMap.forEach(Collections.java:1553)}}{{
> at
> org.apache.kafka.clients.admin.internals.AdminApiFuture$SimpleAdminApiFuture.complete(AdminApiFuture.java:96)}}{{
> at
> org.apache.kafka.clients.admin.internals.AdminApiDriver.complete(AdminApiDriver.java:186)}}{{
> at
> org.apache.kafka.clients.admin.internals.AdminApiDriver.onResponse(AdminApiDriver.java:230)}}{{
> at
> org.apache.kafka.clients.admin.KafkaAdminClient$39.handleResponse(KafkaAdminClient.java:4843)}}{{
> at
> org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.handleResponses(KafkaAdminClient.java:1397)}}{{
> at
> org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1550)}}{{
> at
> org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1473)}}{{{}
> at java.base/java.lang.Thread.run(Thread.java:840){}}}{{{}Caused by:
> org.apache.kafka.common.errors.UnknownTopicOrPartitionException: Failed
> altering consumer group offsets for the following partitions:
> [source.synth-client-topic-0]{}}}{quote}
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
