KIP-382: MirrorMaker 2.0 by ryannedolan · Pull Request #6295 · apache/kafka

ryannedolan · 2019-02-20T16:13:48Z

Implementation of KIP-382 "MirrorMaker 2.0" (approved) and KIP-416 "Notify SourceTask of ACK'd offsets, metadata" (not yet approved).

Depends on #6171

Quick start:

create MM2 configuration file.
launch with ./bin/connect-mirror-maker.sh mm2.properties

Sample configuration file:

clusters = one, two, three, four
one.bootstrap.servers = ...
two.bootstrap.servers = ...
...
one->two.enabled = true # false by default
two->one.enabled = true
...
three->four.topics = topic1, topic2 # .* by default
...
sync.topic.acls.enabled = false # disable for non-secure clusters

The following features of the KIP are deferred for now:
- MirrorSinkConnector/Task -- not used by the MirrorMaker driver, but may be useful to run on a Connect cluster
- "legacy mode" script -- per KIP, not part of first release

…st on build server

connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/Checkpoint.java

connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/ReplicationPolicy.java

connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/RemoteClusterUtils.java

enothereska · 2019-03-14T09:57:32Z

Checked new topic creation in remote cluster
Checked ACL syncing
The code looks good and ideally we get some tests for the above as well. As part of the above check I'm wondering if it makes sense to add an integration test plan to the KIP (apologies if I missed) since some of this functionality needs full e2e testing.

- only run herders for enabled flows - don't proactively create topics -- wait for topic to be created by producer - pluggable filters for topics, groups, config properties - topic-level metrics and other improvements to metrics - drop temporary "monitor" connector

...ct/mirror-client/src/main/java/org/apache/kafka/connect/mirror/DefaultReplicationPolicy.java

connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorCheckpointTask.java

…IP-382

config/connect-mirror-log4j.properties

connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/Checkpoint.java

kujon · 2019-04-05T12:09:56Z

connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/MirrorClient.java

+            .filter(x -> x != -1)
+            .mapToInt(x -> x)
+            .min()
+            .orElse(-1);


I'm wondering: what is the advantage of using -1 over say null to represent no value?

That would work too. Whatever is more conventional in Kafka is fine with me.

There seems to be null some places and other places -1. If possible can we standardize or leave a comment.

connect/mirror/README.md

connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/MirrorClientConfig.java

junrao · 2019-10-03T21:55:41Z

@harshach : There are still a couple of comments that haven't been addressed. The biggest one is on dealing with compacted topic for offset translation.

ryannedolan · 2019-10-03T22:47:52Z

connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceTaskTest.java

+
+        assertTrue("always emit offset sync on first update",
+            partitionState.update(0, 100));
+        assertTrue("upstream offset skipped -> resync",


@junrao holes in a compacted topic are handled, as shown here.

ryannedolan · 2019-10-03T22:53:17Z

@junrao sorry, I didn't see a few comments that were folded/hidden for some reason. Hopefully I've addressed everything that would otherwise delay the merge.

junrao

@ryannedolan : Thanks for the explanation. It all makes sense to me now. So, LGTM.

A few of the limitations with MM2 that I saw (1) if MM2 starts with an existing compact topic with lots of holes, there could be more overhead for writing the offsetSync data. In the worse case, every record requires a checkpoint of offsetSync data. (2) if a consumer starts consuming from the beginning, the offsets won't be translated to the target cluster until the consumer catches up. (3) prefix acls are not propagated. These may not be common issues. However, it would be useful document that in the docs.

Also, we will need a few new message formatters to read the new internal topics. Do you plan to add that before code freeze?

@omkreddy and @harshach : Once the minor issues are addressed, perhaps you could take another look and merge the PR?

junrao · 2019-10-03T22:24:15Z

connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java

+     * @param metadata {@link RecordMetadata} record metadata returned from the broker
+     * @throws InterruptedException
+     */
+    public void commitRecord(SourceRecord record, RecordMetadata metadata)


Could we add some comments to make it clear that one only needs to implement one of the commitRecord()?

Added a couple lines locally. Will hold on to the commit for now -- I don't want to trigger another build at the moment.

...ct/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorConnectorsIntegrationTest.java

omkreddy · 2019-10-04T13:16:35Z

build.gradle

+    testRuntime libs.slf4jlog4j
+  }
+
+  javadoc {


@ryannedolan
We need to include javadocs section for newly added public interfaces/classes.
Example: https://github.com/apache/kafka/blob/trunk/build.gradle#L1539

I assume, we will be adding kafka website documentation as part of KAFKA-8930.
http://kafka.apache.org/documentation/#basic_ops_mirror_maker

Also looks like test failures are related.

Ah thanks @omkreddy I'll fix the javadocs this morning. For the website documentation, we'll need to keep the existing mirror-maker section for now, but I'll add a section re MM2, probably in a separate PR.

The failing tests seem to be related to flakiness in the Connect integration test framework. I'll see what I can do.

omkreddy · 2019-10-04T13:35:09Z

retest this please

omkreddy · 2019-10-05T02:36:22Z

@ryannedolan Any update on test failures?

@harshach Please merge the PR once the Jun's comments are addressed and we have green builds.

ryannedolan · 2019-10-05T04:04:16Z

@omkreddy I traced the build failures to an NPE from KIP-507 committed yesterday. It is breaking MM2's and other Connect integration tests. I'll fix here I guess.

ryannedolan · 2019-10-05T04:18:13Z

I fixed the NPE from KIP-507 -- let's see if we can get a green build now.

ryannedolan · 2019-10-05T12:32:24Z

@harshach good to go!

rhauch

Thanks for updating KIP-416 so that the framework only calls the newer SourceTask.commitRecord(SourceRecord, RecordMetadata).

I have one request to improve the JavaDoc for the new method to make it easier for developers writing their own SourceTask to know whether to override those methods, without having to dive into the Connect runtime to figure that out.

rhauch · 2019-10-05T13:37:45Z

connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java

+     * </p>
+     *
+     * @param record {@link SourceRecord} that was successfully sent via the producer
+     * @param metadata {@link RecordMetadata} record metadata returned from the broker


@ryannedolan, we need to clearly specify that metadata parameter can be null, and then in the JavaDoc above specify what this means for the source task, namely that a transform dropped/skipped the record and it was not written to Kafka.

IMO this is necessary so that developers of connector implementations know what the behavior is so they can properly implement their task. (The JavaDoc was not in KIP-416.)

I also think that it's also worth mentioning here that SourceTask implementations need only implement this method or the older commitRecord(SourceRecord) or neither method, but that generally they do not need to implement both since Connect will only call this method. Again, this will help developers that are implementing their own SourceTask what they need to do.

@rhauch I improved the javadocs further. Should be clear now.

rhauch · 2019-10-05T13:42:25Z

@ryannedolan, @junrao, @harshach: As mentioned above in one of the comments on WorkerSourceTask, this PR also implements KIP-416. I've edited the description of the PR to mention this.

According to the vote thread, KIP-415 has not yet been approved, and per the AK 2.4.0 Release Plan the KIP deadline was on Sept 25.

@junrao, what's required here?

junrao · 2019-10-05T17:10:40Z

@rhauch : We can fold the changes in WorkerSourceTask into this KIP. Ryan, could you update your kip wiki and send an email to the voting thread about the additional changes to make sure that no one objects to this? If kip-416 is completely subsumed by this, we can just cancel it. Otherwise, kip-416 can cover the remaining part.

rhauch · 2019-10-05T17:30:37Z

Thanks, @junrao. KIP-416 is straightforward, limited to the one new method change we discussed (and agreed to), and required for KIP-382, so pulling those changes into KIP-382 makes sense to me.

ijuma · 2019-10-06T15:42:24Z

connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/MirrorClient.java

+     */
+    public Map<TopicPartition, OffsetAndMetadata> remoteConsumerOffsets(String consumerGroupId,
+            String remoteClusterAlias, Duration timeout) {
+        long deadline = System.currentTimeMillis() + timeout.toMillis();


Is there a reason why this is not using the Time interface? We generally never use System.currentTimeMillis() in Kafka.

Good idea, we should replace these in a subsequent PR.

harshach · 2019-10-07T03:58:35Z

@junrao some of the comments mentioned here can be handled in a follow-up patch and can be part of minor release that will follow-up.
Given the no.of users that are interested and the number of reviews, we had it in the PR and also maintaining this big of patch as the trunk continues to evolve will be challenging.
If you don't have any major concerns lets merge this in for 2.4 release and address any new comments in follow-up patch. cc @omkreddy

omkreddy · 2019-10-07T08:21:47Z

@ryannedolan Thanks for the PR. Merging the PR. Lets address any issues in follow-up PRs. Pls raise JIRAs for any pending work (MirrorSinkConnector, legacy mode etc.) for next releases.

amanullah92 · 2020-02-20T09:06:17Z

connect/mirror/README.md

+blacklisted ones), across all enabled replication flows. Each
+replication flow must be explicitly enabled to begin replication:
+
+    A->B.enabled = true


This directive (A->B.enabled) is missing in KIP in the "Running MirrorMaker in production section"-- I got the cluster up but no replication was happening. Until I saw this and fixed this. I am new to MIrrorMaker/Connect Framework- forgive me if this is a well known thing.

bpux · 2020-03-25T18:02:27Z

@ryannedolan try to find JIRA of MirrorSinkConnector but search return nothing. can you please point me where i can find status of this work? thanks!

ewencp · 2020-03-25T18:08:29Z

@bpux https://issues.apache.org/jira/browse/KAFKA-7500, the primary JIRA for KIPs are linked from the KIP document itself: https://cwiki.apache.org/confluence/display/KAFKA/KIP-382%3A+MirrorMaker+2.0

bpux · 2020-03-25T18:15:44Z

@ewencp thanks! will comment in there

michael-trelinski · 2020-04-28T01:13:39Z

Hi @ryannedolan -

I just got done setting up Mirror Maker V2 where I work. It was successful, but the documentation doesn't match the code in various places.

A couple of things that "got me" -

The JMX Metrics in the documentation:

The mbean name for these metrics will be: kafka.mirror.connect:type=MirrorSourceConnect,target=([.\w]+),topic=([.\w]+),partition=([.\d]+) and kafka.mirror.connect:type=MirrorCheckpointConnector,target=([.\w]+),source=([.\w]+),group=([.\w]+)

It should actually be something like (pardon me, I'm using the JMX-prometheus exporter):
kafka.connect.mirror<type=MirrorSourceConnector, target=([\-\w]+), topic=([\-\w]+), partition=(\d+)>. Note how it's actually kafka.connect.mirror not kafka.mirror.connect, also it's MirrorSourceConnector, not MirrorSourceConnect.

In the documentation, there is a part about blacklisting groups:

groups.blacklist | empty string | groups to exclude from replication

In the code, it is public static final String GROUPS_BLACKLIST_DEFAULT = "console-consumer-.*, connect-.*, __.*";; here: https://github.com/apache/kafka/blob/trunk/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/DefaultGroupFilter.java#L36 - this got me because we're actually using some Kafka Connect stuff for other projects, but we're running Mirror Maker as it's own cluster. Maybe you could blacklist only Mirror Maker's Kafka Connect connect- consumer group?

This one is kind of a nit-pick, but in RemoteClusterUtils.translateOffsets, the timeout parameter isn't really a timeout in the conventional sense. Stepping further into the call that's made using timeout, return client.remoteConsumerOffsets(consumerGroupId, remoteClusterAlias, timeout); we can see that it's being used as a deadline, not a timeout. In MirrorClient's remoteConsumerOffsets method, it's being used as a timeout, and as a deadline. Setting this too low caused me to miss some offsets because I didn't understand how it was being used. Setting this absurdly high got me the results that I wanted.

All in all a great product and vastly superior to Mirror Maker V1. Thanks!

Migueljfs · 2022-01-03T16:33:16Z

I can't seem to find the new metrics mentioned in the KIP.
Like @michael-trelinski said above, I've tried both kafka.connect.mirror / kafka.mirror.connect AND MirrorSourceConnector / MirrorSourceConnect and all combinations of them but still I don't see the mirror domain the respective mbeans.

I installed JMXTERM on my mirror-maker pod and this is what I got:

Domains:

$>domains
#following domains are available
JMImplementation
com.sun.management
java.lang
java.nio
java.util.logging
jdk.management.jfr
kafka.connect
kafka.consumer
kafka.producer

Beans for connect:

#domain = kafka.connect:
kafka.connect:client-id=connect-1,node-id=node--1,type=connect-node-metrics
kafka.connect:client-id=connect-1,node-id=node-7,type=connect-node-metrics
kafka.connect:client-id=connect-1,type=app-info
kafka.connect:client-id=connect-1,type=connect-coordinator-metrics
kafka.connect:client-id=connect-1,type=connect-metrics
kafka.connect:client-id=connect-1,type=kafka-metrics-count
kafka.connect:id="west->central",type=app-info
kafka.connect:id=connect-1,type=app-info
kafka.connect:type=app-info
kafka.connect:type=connect-worker-metrics
kafka.connect:type=connect-worker-rebalance-metrics
kafka.connect:type=kafka-metrics-count
#domain = kafka.consumer:

As you can see, no mentions of "mirror" of any kind anywhere.

Anyone able to help me understand what's going on?

ryannedolan added 5 commits February 20, 2019 00:30

KIP-382: MirrorMaker 2.0

524ad2d

:connect:mirror and :connect:mirror-client were missing IntegrationTe…

ca086bb

…st on build server

drop unnecessary change to server.properties

4f39ad1

fixed SSL/SASL support

cb18c36

default to a meaningful producer client.id

d7b6c0e

enothereska reviewed Mar 7, 2019

View reviewed changes

ryannedolan added 3 commits March 11, 2019 19:17

topic metrics. pluggable filters. inherit connector configs.

016c0ad

don't proactively create topics

e1fd046

support tasks.max, topics.blacklist etc in top-level MirrorMakerConfig

83d9125

ryannedolan added 2 commits March 14, 2019 13:14

only run herders for enabled flows

9d3eb78

williamhammond reviewed Mar 18, 2019

View reviewed changes

...ct/mirror-client/src/main/java/org/apache/kafka/connect/mirror/DefaultReplicationPolicy.java Show resolved Hide resolved

williamhammond reviewed Mar 28, 2019

View reviewed changes

connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorCheckpointTask.java Outdated Show resolved Hide resolved

ryannedolan added 4 commits April 3, 2019 12:00

simpler metrics. bug fixes.

cf0ae71

bug fix

d648275

bug fix

82295d8

Merge branch 'KIP-382' of https://github.com/ryannedolan/kafka into K…

b74934c

…IP-382

viktorsomogyi reviewed Apr 4, 2019

View reviewed changes

config/connect-mirror-log4j.properties Outdated Show resolved Hide resolved

connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/Checkpoint.java Show resolved Hide resolved

ryannedolan added 2 commits April 4, 2019 11:47

fix indentation

685c0bc

fixed race conditions on stop()

a4b8a38

kujon reviewed Apr 5, 2019

View reviewed changes

ryannedolan commented Apr 10, 2019

View reviewed changes

connect/mirror/README.md Outdated Show resolved Hide resolved

halorgium reviewed Apr 11, 2019

View reviewed changes

connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/MirrorClientConfig.java Show resolved Hide resolved

ryannedolan added 6 commits April 12, 2019 15:09

proactively create topics, partitions

e92a6c3

checkstyle typo

2909d1f

lag -> latency when referring to time

76eb983

checkpoint metrics and sourceClusters()

132a6cd

upstreamTopics()

572e63c

drop upstreamTopics and sourceClusters; expose ReplicationPolicy instead

d0cc810

ryannedolan commented Oct 3, 2019

View reviewed changes

junrao approved these changes Oct 4, 2019

View reviewed changes

omkreddy reviewed Oct 4, 2019

View reviewed changes

enable javadocs

c099465

ryannedolan added 2 commits October 4, 2019 23:14

drop extraneous throws

ac75bcf

fix NPE in DistributedHerder -- unrelated to KIP-382

e823ad4

rhauch reviewed Oct 5, 2019

View reviewed changes

improve javadocs. combine integration tests into one test.

b6688cc

ijuma reviewed Oct 6, 2019

View reviewed changes

omkreddy closed this in 4ac892c Oct 7, 2019

ryannedolan deleted the KIP-382 branch December 24, 2019 00:14

amanullah92 reviewed Feb 20, 2020

View reviewed changes

C0urante mentioned this pull request Nov 8, 2022

KAFKA-14021: Implement new KIP-618 APIs in MirrorSourceConnector #12366

Merged

3 tasks

Conversation

ryannedolan commented Feb 20, 2019 • edited by omkreddy Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

enothereska commented Mar 14, 2019

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

junrao commented Oct 3, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ryannedolan commented Oct 3, 2019

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

omkreddy commented Oct 4, 2019

Uh oh!

omkreddy commented Oct 5, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ryannedolan commented Oct 5, 2019

Uh oh!

ryannedolan commented Oct 5, 2019

Uh oh!

ryannedolan commented Oct 5, 2019

Uh oh!

rhauch left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rhauch commented Oct 5, 2019

Uh oh!

junrao commented Oct 5, 2019

Uh oh!

rhauch commented Oct 5, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

harshach commented Oct 7, 2019

Uh oh!

omkreddy commented Oct 7, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bpux commented Mar 25, 2020

Uh oh!

ewencp commented Mar 25, 2020

Uh oh!

bpux commented Mar 25, 2020

Uh oh!

michael-trelinski commented Apr 28, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

ryannedolan commented Feb 20, 2019 •

edited by omkreddy

Loading

omkreddy commented Oct 5, 2019 •

edited

Loading

michael-trelinski commented Apr 28, 2020 •

edited

Loading