OmniaGM commented on code in PR #15999:
URL: https://github.com/apache/kafka/pull/15999#discussion_r1643099652
##########
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceTask.java:
##########
@@ -52,57 +48,47 @@ public class MirrorSourceTask extends SourceTask {
private static final Logger log =
LoggerFactory.getLogger(MirrorSourceTask.class);
- private static final int MAX_OUTSTANDING_OFFSET_SYNCS = 10;
-
private KafkaConsumer<byte[], byte[]> consumer;
- private KafkaProducer<byte[], byte[]> offsetProducer;
private String sourceClusterAlias;
- private String offsetSyncsTopic;
private Duration pollTimeout;
- private long maxOffsetLag;
private Map<TopicPartition, PartitionState> partitionStates;
private ReplicationPolicy replicationPolicy;
private MirrorSourceMetrics metrics;
private boolean stopping = false;
- private final Map<TopicPartition, OffsetSync> delayedOffsetSyncs = new
LinkedHashMap<>();
- private final Map<TopicPartition, OffsetSync> pendingOffsetSyncs = new
LinkedHashMap<>();
- private Semaphore outstandingOffsetSyncs;
private Semaphore consumerAccess;
+ private OffsetSyncWriter offsetSyncWriter;
public MirrorSourceTask() {}
// for testing
MirrorSourceTask(KafkaConsumer<byte[], byte[]> consumer,
MirrorSourceMetrics metrics, String sourceClusterAlias,
ReplicationPolicy replicationPolicy, long maxOffsetLag,
KafkaProducer<byte[], byte[]> producer,
Semaphore outstandingOffsetSyncs, Map<TopicPartition,
PartitionState> partitionStates,
- String offsetSyncsTopic) {
+ String offsetSyncsTopic, boolean emitOffsetSyncEnabled) {
Review Comment:
updated the constructor to accept OffsetSyncWriter instead
--
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
