KAFKA-16254: Allow MM2 to fully disable offset sync feature by OmniaGM · Pull Request #15999 · apache/kafka

OmniaGM · 2024-05-20T14:28:57Z

Implementation of KIP-1031 https://cwiki.apache.org/confluence/display/KAFKA/KIP-1031%3A+Control+offset+translation+in+MirrorSourceConnector

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

chia7712

@OmniaGM I have left some comments in the discussion thread (https://lists.apache.org/thread/2scgvn66vs8c04ldxsj4sw2vlo6wd98o). It would be great if you have free time to take a look.

For another code style idea: Could we have a new static class including all "emit"-related code? That will have following benefits.

more readable. The static class can be null (or empty) to "prove" that this function is "disabled"
no unused objects are created. For example, offsetProducer, delayedOffsetSyncs, pendingOffsetSyncs are unused if the emit is disable
easy to test emit behavior as all they in single static class now.

WDYT?

OmniaGM · 2024-05-21T17:03:54Z

@OmniaGM I have left some comments in the discussion thread (https://lists.apache.org/thread/2scgvn66vs8c04ldxsj4sw2vlo6wd98o). It would be great if you have free time to take a look.

For another code style idea: Could we have a new static class including all "emit"-related code? That will have following benefits.

more readable. The static class can be null (or empty) to "prove" that this function is "disabled"

no unused objects are created. For example, offsetProducer, delayedOffsetSyncs, pendingOffsetSyncs are unused if the emit is disable

easy to test emit behavior as all they in single static class now.

WDYT?

We kinda have OffsetSyncStore and based on the name it looks like the right place for this however, it is focused on read from the offsetSyncs topic. I added another one called OffsetSyncWriter and moved all emitting offsets code into there. let me know if this is okay!

chia7712

@OmniaGM Nice patch!

soarez

Thanks for this PR @OmniaGM

…llow-offset-sync

chia7712 · 2024-05-29T20:50:09Z

just remind that today is the deadline for feature freeze. If this PR is necessary for 3.8.0, we need to cherry-pick to branch-3.8

OmniaGM · 2024-05-29T21:41:17Z

I was hoping for it to land in 3.8.0 I'll raise a PR against 3.8 once we merge it into trunk. however It shouldn't be a blocker for 3.8

mimaison

Thanks for the PR! I left a few comments.

…llow-offset-sync

OmniaGM · 2024-06-28T22:45:11Z

@C0urante I believe I addressed all the comments now, can you please have a second look?

C0urante

Thanks, this is looking really good! Had some more thoughts on the validation logic but apart from that I think we're basically there. The integration tests (minus the accidental failed creation for the MirrorCheckpointConnector) look especially nice 👍

C0urante

Thanks @OmniaGM! Really close, made it all the way through the tests and the comments from this round should be all that need to be addressed before we can merge.

C0urante

LGTM, thanks @OmniaGM!

C0urante · 2024-07-02T02:05:17Z

@mimaison @chia7712 @soarez I plan on merging this by the end of the week; if any of you would like to do another round before I merge, please let me know.

mimaison

Thanks for the updates. I made another pass. It looks good, I just left a few small suggestions.

mimaison

LGTM

chia7712

@OmniaGM thanks for your patch and patient. a couple of comments left PTAL

chia7712 · 2024-07-02T17:34:08Z

+        List<ConfigValue> configValues = super.validate(connectorConfigs).configValues();
+        MirrorCheckpointConfig.validate(connectorConfigs).forEach((config, errorMsg) ->
+                configValues.stream()
+                        .filter(conf -> conf.name().equals(config))


EMIT_OFFSET_SYNCS_ENABLED is in MirrorSourceConfig, so it is not a part of config def of MirrorCheckpointConnector. Hence, the error related to EMIT_OFFSET_SYNCS_ENABLED can't be propagated.

The following test shows my concern:

@Test public void test() { Map<String, String> props = new HashMap<>(); props.put(MirrorConnectorConfig.EMIT_OFFSET_SYNCS_ENABLED, "false"); MirrorCheckpointConnector connector = new MirrorCheckpointConnector(); Config config = connector.validate(props); assertEquals(1, config.configValues().stream().filter(c -> c.name().equals(MirrorConnectorConfig.EMIT_OFFSET_SYNCS_ENABLED)).count()); }

I moved this to MirrorConnectorConfig which all inherit. So this should be addressed now

I still agree with @mimaison that we shouldn't have this property in the MirrorConnectorConfig class, since then it'll be included as a "common config" for all connectors in our generated docs here instead of in the MirrorSourceConnector-specific docs here.

Can we just add a new ConfigValue for the property in MirrorCheckpointConnector::validate if we find a validation error with it?

I updated MirrorCheckpointConnector::validate to create configValue if the name of the config not defined

Sorry for the confusion I just lost track of @mimaison point from before! this why I reverted config back to MirrorConnectorConfig

No worries! Latest looks good to me 👍

@chia7712 thoughts?

chia7712 · 2024-07-02T17:45:01Z

+        boolean offsetSyncsConfigured = props.keySet().stream()
+                .anyMatch(conf -> conf.startsWith(OFFSET_SYNCS_CLIENT_ROLE_PREFIX) || conf.startsWith(OFFSET_SYNCS_TOPIC_CONFIG_PREFIX));
+
+        if ("false".equals(props.get(MirrorConnectorConfig.EMIT_OFFSET_SYNCS_ENABLED)) && offsetSyncsConfigured) {


what if EMIT_OFFSET_SYNCS_ENABLED=true and offsetSyncsConfigured=false? Should we add error message for it?

No, because offsetSyncsConfigured has default which is the global configs for topics and clients

chia7712

@OmniaGM thanks for updated PR. the last two comments are left. PTAL

chia7712 · 2024-07-03T20:30:32Z

+
+        if (emitCheckpointDisabled && syncGroupOffsetsDisabled) {
+            Arrays.asList(SYNC_GROUP_OFFSETS_ENABLED, EMIT_CHECKPOINTS_ENABLED).forEach(configName -> {
+                invalidConfigs.putIfAbsent(configName, "MirrorCheckpointConnector can't run with both " + SYNC_GROUP_OFFSETS_ENABLED + " and " +


It seems to me EMIT_CHECKPOINTS_ENABLED does not obstruct MirrorCheckpointConnector from running since it is used to update consumer groups offsets of target cluster. By contrast, SYNC_GROUP_OFFSETS_ENABLED do impact the MirrorCheckpointConnector

https://github.com/apache/kafka/blob/trunk/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorCheckpointConnector.java#L121

My understanding that it will not start a task so the connector would be doing nothing without any feedback or indication for why. If this is problematic I would suggest remove this check as this was a suggestion out of the scope of the KIP to begin with. @C0urante and @chia7712 WDYT?

I didn't realize that we disabled the connector from generating task configs if checkpoints were disabled. In that case, I think we should remove the check for sync.group.offsets.enabled altogether and just check emit.checkpoints.enabled. But generally I agree with @OmniaGM that if we don't generate task configs for the connector when it's configured that way, it's best to surface this as a validation error instead of silently degrading to a no-op.

It seems to me EMIT_CHECKPOINTS_ENABLED does not obstruct MirrorCheckpointConnector from running since it is used to update consumer groups offsets of target cluster. By contrast, SYNC_GROUP_OFFSETS_ENABLED do impact the MirrorCheckpointConnector

I just notice that I use the incorrect config name :(

My understanding that it will not start a task so the connector would be doing nothing without any feedback or indication for why.

agree and my point was "we should return invalid config if EMIT_CHECKPOINTS_ENABLED=false", because SYNC_GROUP_OFFSETS_ENABLED=false is allowed to MirrorCheckpointConnector . For example: in this case: EMIT_CHECKPOINTS_ENABLED=false and SYNC_GROUP_OFFSETS_ENABLED=true we should return invalid configs, right?

I removed the validation for sync.group.offsets.enabled and kept emit.checkpoints.enabled validation only

chia7712

LGTM and thanks @OmniaGM patient to complete this great feature!

C0urante

Ditto @chia7712's comments; thank you so much for your patience @OmniaGM!

OmniaGM · 2024-07-09T09:23:45Z

Thanks All for reviewing this PR. I reviewed the failed tests and they are unrelated.

…5999) Reviewers: Mickael Maison <mickael.maison@gmail.com>, Chris Egerton <fearthecellos@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>, Igor Soarez <soarez@apple.com>

gharris1727 added mirror-maker-2 kip Requires or implements a KIP labels May 20, 2024

chia7712 reviewed May 20, 2024

View reviewed changes

Comment thread connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceTask.java Outdated

OmniaGM added 2 commits May 21, 2024 15:26

KAFKA-16254: Allow MM2 to fully disable offset sync feature

6ca1471

address part of the feedback

3038580

OmniaGM force-pushed the KAFKA-16254-mm2-disallow-offset-sync branch from 5ae6390 to 3038580 Compare May 21, 2024 15:26

refactor

485c3db

Add OffsetSyncWriter

557591b

OmniaGM force-pushed the KAFKA-16254-mm2-disallow-offset-sync branch from 10eb256 to 557591b Compare May 21, 2024 17:06

chia7712 reviewed May 21, 2024

View reviewed changes

address feedback

4958ac1

OmniaGM force-pushed the KAFKA-16254-mm2-disallow-offset-sync branch from 436ce41 to 4958ac1 Compare May 22, 2024 09:38

soarez reviewed May 23, 2024

View reviewed changes

OmniaGM added 3 commits May 28, 2024 12:40

address feedback

69823ae

Merge remote-tracking branch 'apache/trunk' into KAFKA-16254-mm2-disa…

875fcdb

…llow-offset-sync

address igor feedback

27c0754

OmniaGM requested a review from soarez May 29, 2024 16:44

soarez reviewed May 29, 2024

View reviewed changes

mimaison reviewed May 30, 2024

View reviewed changes

OmniaGM added 5 commits June 4, 2024 19:17

address feedback

51c25c8

address feedback

7776618

Merge remote-tracking branch 'apache/trunk' into KAFKA-16254-mm2-disa…

3927e6f

…llow-offset-sync

feedback

a3fd873

feedback

b687bbd

OmniaGM added 5 commits June 27, 2024 10:54

Apply spotless fixes

97de142

address feedback

db3cb92

move PartitionState to OffsetSyncWriter

bad7954

add suggested comments

3d14a17

Fail checkpoint connector when config is invalid

00c2cbd

C0urante reviewed Jun 29, 2024

View reviewed changes

OmniaGM added 2 commits July 1, 2024 19:03

address feedback

a0b8ab5

fix spotless

dabcaa7

C0urante reviewed Jul 1, 2024

View reviewed changes

address feedback

fd6e0e0

OmniaGM force-pushed the KAFKA-16254-mm2-disallow-offset-sync branch from 66e0246 to fd6e0e0 Compare July 1, 2024 22:26

address feedback

911e6e8

C0urante approved these changes Jul 2, 2024

View reviewed changes

mimaison reviewed Jul 2, 2024

View reviewed changes

OmniaGM added 2 commits July 2, 2024 11:32

address feedback

5f0f2d6

add spaces

d468c75

mimaison approved these changes Jul 2, 2024

View reviewed changes

chia7712 reviewed Jul 2, 2024

View reviewed changes

address chia's feedback

9f449a9

OmniaGM requested a review from chia7712 July 3, 2024 13:47

Create ConfigValue

cda08cd

chia7712 reviewed Jul 3, 2024

View reviewed changes

feedback

6deb95c

chia7712 approved these changes Jul 8, 2024

View reviewed changes

C0urante approved these changes Jul 8, 2024

View reviewed changes

mimaison merged commit 0781151 into apache:trunk Jul 9, 2024

Conversation

OmniaGM commented May 20, 2024

Committer Checklist (excluded from commit message)

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

OmniaGM commented May 21, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

soarez left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

chia7712 commented May 29, 2024

Uh oh!

OmniaGM commented May 29, 2024

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

mimaison left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

OmniaGM commented Jun 28, 2024

Uh oh!

C0urante left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

C0urante left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

C0urante left a comment

Choose a reason for hiding this comment

Uh oh!

C0urante commented Jul 2, 2024

Uh oh!

mimaison left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

OmniaGM commented May 21, 2024 •

edited

Loading