C0urante commented on code in PR #13178: URL: https://github.com/apache/kafka/pull/13178#discussion_r1106023715
########## connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorCheckpointTask.java: ########## @@ -134,9 +138,9 @@ public String version() { @Override public List<SourceRecord> poll() throws InterruptedException { try { - long deadline = System.currentTimeMillis() + interval.toMillis(); - while (!stopping && System.currentTimeMillis() < deadline) { - offsetSyncStore.update(pollTimeout); + if (stopping.await(pollTimeout.toMillis(), TimeUnit.MILLISECONDS)) { Review Comment: Also, we changed the contract for `SourceTask::stop` a while ago to not be invoked on a separate thread from the one that invokes `SourceTask::poll` in order to fix KAFKA-10792, so I'm not sure how much value changing this to a `CountDownLatch` really adds here. Unless there's something else I'm missing here that necessitates this change, I think it's fine to leave `stopping` as a boolean field. As a follow-up, we could potentially tweak the logic here to pause for, e.g., a second and then return a `null` record batch from `SourceTask::poll` if the poll timeout hasn't elapsed since the last time we returned a non-null batch. But that should not happen in this PR, we're doing plenty here already. ########## connect/mirror/src/main/java/org/apache/kafka/connect/mirror/OffsetSyncStore.java: ########## @@ -75,18 +123,9 @@ OptionalLong translateDownstream(TopicPartition sourceTopicPartition, long upstr } } - // poll and handle records - synchronized void update(Duration pollTimeout) { - try { - consumer.poll(pollTimeout).forEach(this::handleRecord); - } catch (WakeupException e) { - // swallow - } - } - - public synchronized void close() { - consumer.wakeup(); - Utils.closeQuietly(consumer, "offset sync store consumer"); + public void close() { Review Comment: Nit: ```suggestion @Override public void close() { ``` ########## connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorCheckpointTask.java: ########## @@ -134,9 +138,9 @@ public String version() { @Override public List<SourceRecord> poll() throws InterruptedException { try { - long deadline = System.currentTimeMillis() + interval.toMillis(); - while (!stopping && System.currentTimeMillis() < deadline) { - offsetSyncStore.update(pollTimeout); + if (stopping.await(pollTimeout.toMillis(), TimeUnit.MILLISECONDS)) { Review Comment: Should this condition be inverted? ########## connect/mirror/src/main/java/org/apache/kafka/connect/mirror/OffsetSyncStore.java: ########## @@ -16,40 +16,88 @@ */ package org.apache.kafka.connect.mirror; -import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.common.TopicPartition; -import org.apache.kafka.common.errors.WakeupException; -import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Utils; +import org.apache.kafka.connect.util.KafkaBasedLog; +import org.apache.kafka.connect.util.TopicAdmin; -import java.util.Map; -import java.util.HashMap; import java.util.Collections; -import java.time.Duration; +import java.util.Map; import java.util.Optional; import java.util.OptionalLong; +import java.util.concurrent.ConcurrentHashMap; /** Used internally by MirrorMaker. Stores offset syncs and performs offset translation. */ class OffsetSyncStore implements AutoCloseable { - private final KafkaConsumer<byte[], byte[]> consumer; - private final Map<TopicPartition, OffsetSync> offsetSyncs = new HashMap<>(); - private final TopicPartition offsetSyncTopicPartition; + private final KafkaBasedLog<byte[], byte[]> backingStore; + private final Map<TopicPartition, OffsetSync> offsetSyncs = new ConcurrentHashMap<>(); + private final TopicAdmin admin; + private volatile boolean readToEnd = false; OffsetSyncStore(MirrorCheckpointConfig config) { - consumer = new KafkaConsumer<>(config.offsetSyncsTopicConsumerConfig(), - new ByteArrayDeserializer(), new ByteArrayDeserializer()); - offsetSyncTopicPartition = new TopicPartition(config.offsetSyncsTopic(), 0); - consumer.assign(Collections.singleton(offsetSyncTopicPartition)); + Consumer<byte[], byte[]> consumer = null; + TopicAdmin admin = null; + KafkaBasedLog<byte[], byte[]> store; + try { + Consumer<byte[], byte[]> finalConsumer = consumer = MirrorUtils.newConsumer(config.offsetSyncsTopicConsumerConfig()); Review Comment: This is fairly unclean but I can't think of a significantly better alternative... guess it's good enough for now. ########## connect/mirror/src/main/java/org/apache/kafka/connect/mirror/OffsetSyncStore.java: ########## @@ -16,40 +16,88 @@ */ package org.apache.kafka.connect.mirror; -import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.admin.AdminClientConfig; +import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.common.TopicPartition; -import org.apache.kafka.common.errors.WakeupException; -import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Utils; +import org.apache.kafka.connect.util.KafkaBasedLog; +import org.apache.kafka.connect.util.TopicAdmin; -import java.util.Map; -import java.util.HashMap; import java.util.Collections; -import java.time.Duration; +import java.util.Map; import java.util.Optional; import java.util.OptionalLong; +import java.util.concurrent.ConcurrentHashMap; /** Used internally by MirrorMaker. Stores offset syncs and performs offset translation. */ class OffsetSyncStore implements AutoCloseable { - private final KafkaConsumer<byte[], byte[]> consumer; - private final Map<TopicPartition, OffsetSync> offsetSyncs = new HashMap<>(); - private final TopicPartition offsetSyncTopicPartition; + private final KafkaBasedLog<byte[], byte[]> backingStore; + private final Map<TopicPartition, OffsetSync> offsetSyncs = new ConcurrentHashMap<>(); + private final TopicAdmin admin; + private volatile boolean readToEnd = false; OffsetSyncStore(MirrorCheckpointConfig config) { - consumer = new KafkaConsumer<>(config.offsetSyncsTopicConsumerConfig(), - new ByteArrayDeserializer(), new ByteArrayDeserializer()); - offsetSyncTopicPartition = new TopicPartition(config.offsetSyncsTopic(), 0); - consumer.assign(Collections.singleton(offsetSyncTopicPartition)); + Consumer<byte[], byte[]> consumer = null; + TopicAdmin admin = null; + KafkaBasedLog<byte[], byte[]> store; + try { + Consumer<byte[], byte[]> finalConsumer = consumer = MirrorUtils.newConsumer(config.offsetSyncsTopicConsumerConfig()); + TopicAdmin finalAdmin = admin = new TopicAdmin( + config.offsetSyncsTopicAdminConfig().get(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG), + config.forwardingAdmin(config.offsetSyncsTopicAdminConfig())); + store = new KafkaBasedLog<byte[], byte[]>( + config.offsetSyncsTopic(), + Collections.emptyMap(), + Collections.emptyMap(), + () -> finalAdmin, + (error, record) -> this.handleRecord(record), + Time.SYSTEM, + ignored -> { + }) { + + @Override + protected Producer<byte[], byte[]> createProducer() { + return null; + } + + @Override + protected Consumer<byte[], byte[]> createConsumer() { + return finalConsumer; + } + + @Override + protected boolean readPartition(TopicPartition topicPartition) { + return topicPartition.partition() == 0; + } + }; + } catch (Throwable t) { + Utils.closeQuietly(consumer, "consumer for offset syncs"); + Utils.closeQuietly(admin, "admin client for offset syncs"); + throw t; + } + this.admin = admin; + this.backingStore = store; } - // for testing - OffsetSyncStore(KafkaConsumer<byte[], byte[]> consumer, TopicPartition offsetSyncTopicPartition) { - this.consumer = consumer; - this.offsetSyncTopicPartition = offsetSyncTopicPartition; + OffsetSyncStore() { + this.admin = null; + this.backingStore = null; + } + + public void start() { Review Comment: Can we add a Javadoc here? At a minimum, we should make it clear that this performs a synchronous read to the end of the topic. ########## connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java: ########## @@ -392,6 +401,10 @@ protected Consumer<K, V> createConsumer() { return new KafkaConsumer<>(consumerConfigs); } + protected boolean readPartition(TopicPartition topicPartition) { Review Comment: Nit: add a Javadoc explaining when this method will be used, what it will be used for, and that it's designed to be overridden by subclasses? ########## connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java: ########## @@ -257,8 +257,17 @@ public void start() { " allotted period. This could indicate a connectivity issue, unavailable topic partitions, or if" + " this is your first use of the topic it may have taken too long to create."); - for (PartitionInfo partition : partitionInfos) - partitions.add(new TopicPartition(partition.topic(), partition.partition())); + for (PartitionInfo partition : partitionInfos) { + TopicPartition topicPartition = new TopicPartition(partition.topic(), partition.partition()); + if (readPartition(topicPartition)) { + partitions.add(topicPartition); + } + } + if (partitions.isEmpty()) { + throw new ConnectException("Some partitions for " + topic + " exist, but no partitions matched the " + + "required filter. This could indicate a connectivity issue, unavailable topic partitions, or if" + + " this is your first use of the topic it may have taken too long to create."); Review Comment: What circumstances do we realistically anticipate that would lead to this which involve connectivity issues, unavailable topic partitions, or a lag in topic metadata propagation after it has been created? IMO it's fine to keep just the first sentence here. -- 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