KAFKA-4794: Add access to OffsetStorageReader from SourceConnector by fhussonnois · Pull Request #2604 · apache/kafka

fhussonnois · 2017-02-27T22:40:28Z

This a first attempt to implement Add access to OffsetStorageReader from Source Connector.
I am not sure if I did it right so I prefer to ask you your feedback. I still need to write some tests.

asfbot · 2017-02-27T23:24:52Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1867/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-02-27T23:25:20Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1864/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-02-27T23:53:14Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1865/
Test PASSed (JDK 8 and Scala 2.12).

hleb-albau · 2017-08-24T16:24:27Z

Will this path be merged?

fhussonnois · 2017-08-31T09:51:23Z

Hi @rhauch, this PR has been updated regarding the KIP-131

rhauch

Looks great, @fhussonnois! I do have a few specific requests, but all are pretty minor.

tedyu · 2017-09-05T16:41:59Z

Is it possible to trigger infinite loop:

raiseError -> reconfig -> raiseError -> reconfig ...

fhussonnois · 2017-09-05T18:59:54Z

Thank you @rhauch for your comments. I've updated the PR.

fhussonnois · 2017-09-07T19:42:44Z

@rhauch is there anything I should add or modify for the PR ?

rhauch · 2017-09-07T19:52:20Z

@fhussonnois, I just submitted the KIP for a vote, which will be open for a few days. This can be merged once that is approved.

frankcoutinho · 2020-03-05T18:18:45Z

I also needed this feature. #feelsbad

rhauch · 2020-03-05T19:10:34Z

@fhussonnois, would you mind updating this PR to eliminate the conflicts?

Also, while I hope to get this into AK 2.6, the KIP hasn't yet passed. I did make a suggestion on the vote thread, though, to consider making it much easier for connectors that want to also run in older versions of Connect to easily get the OffsetStorageReader when Connect supports it without jumping through a lot of ugly hoops.

rhauch · 2020-05-06T21:08:03Z

ok to test

rhauch

Thanks, @fhussonnois. I have a few more suggestions to clean up the implementation a bit more, to improve backward compatibility, and to ensure the new code is adequately tested.

rhauch · 2020-05-07T19:28:12Z

At one point these were anonymous classes, and the delegation perhaps made a bit more sense. Now that they are inner classes, it seems like it would be a lot less confusing and simpler if the DelegateToWorkerConnectorContext class were extended by the DelegateSinkConnectorContext and DelegateSourceConnectorContext classes. Doing this has several advantages:

removes nesting/delegation and eliminates the chance of getting into an infinite loop

eliminates duplicate code in these classes

simplifies the context classes quite a bit

makes it more obvious that the sink connector context has nothing extra over the base

makes it more obvious that the source connector context only adds the offset storage reader

simplifies this code a bit (see below)

By keeping DelegateToWorkerConnectorContext class non-abstract, we're actually able to maintain backward compatibility by calling initialize when the connector class is neither a source or sink connector:

This code becomes simpler, too:

Suggested change

final ConnectorContext delegateCtx = new DelegateToWorkerConnectorContext();

if (isSinkConnector()) {

SinkConnectorConfig.validate(config);

connector.initialize(new DelegateSinkConnectorContext(delegateCtx));

} else {

connector.initialize(new DelegateSourceConnectorContext(delegateCtx, offsetStorageReader));

}

if (isSinkConnector()) {

SinkConnectorConfig.validate(config);

connector.initialize(new DelegateSinkConnectorContext());

} else if (isSourceConnector()) {

connector.initialize(new DelegateSourceConnectorContext(offsetStorageReader));

} else {

connector.initialize(new DelegateToWorkerConnectorContext());

}

This seems a little strange, but we're actually not checking whether the Connector instance passed into this WorkerConnector is only an instance of SourceConnector or SinkConnector. If it is neither, prior to this change the framework would still initialize it, and I think we should maintain that arguably strange behavior just to maintain backward compatibility.

rhauch · 2020-05-07T19:28:50Z

Nit: let's avoid unrelated line additions.

rhauch · 2020-05-07T19:29:03Z

Nit: let's avoid unrelated line additions.

rhauch · 2020-05-21T23:14:13Z

retest this please

rhauch · 2020-05-21T23:14:28Z

ok to test

Added commits to make the requested changes

rhauch

@fhussonnois, thanks again for doing the brunt of the work on this feature. We're getting close to AK 2.6.0 feature freeze, and it'd be great to merge this feature now that KIP-131 has been adopted. Since my previous review was some time ago, I hope you don't mind that I pushed the remaining changes I suggested in my previous review. I've also dismissed my previous "Request Changes" review.

@kkonstantine would you mind taking a look? I'd like to have an independent review of my changes. Thanks!

fhussonnois · 2020-05-21T23:28:16Z

@rhauch thank you very much for finalizing this PR; I apologize for not finding any free time to work on it these past few weeks.

rhauch · 2020-05-23T14:05:29Z

retest this please

rhauch · 2020-05-24T13:49:46Z

Just one failure in each of the 3 builds, and none were in Connect unit or integration tests.

kkonstantine

Thanks for the original contribution and the KIP @fhussonnois

@rhauch thanks for completing this PR. Overall looks good. I have a few minor comments and then I think that maybe we shouldn't plan for a connector implementation that is not a sink or a source in the tests. The code would break elsewhere anyways, and this is not supported by they framework today. Wdyt?

Add two interfaces SinkConnectorContext/SourceConnectContext that extend ConnectorContext in order to expose an OffsetStorageReader instance.

…initialization in WorkerConnector

…nector default methods

…ce or sink, and incorporated other feedback

rhauch · 2020-05-24T16:52:45Z

Rebased to correct conflicts.

rhauch · 2020-05-24T16:53:01Z

ok to test

rhauch · 2020-05-24T16:53:50Z

retest this please

rhauch · 2020-05-24T19:29:17Z

All 3 builds had test failures on known flaky Streams integration tests, but all Connect-related tests passed.

kkonstantine

This looks great now. Happy to see this feature being added!
Thanks @rhauch
LGTM

fhussonnois force-pushed the KAFKA-4794 branch from e2bb682 to 9f38088 Compare August 31, 2017 09:48

rhauch previously requested changes Sep 5, 2017

View reviewed changes

tedyu reviewed Sep 5, 2017

View reviewed changes

Comment thread connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java Outdated

tedyu reviewed Sep 5, 2017

View reviewed changes

fhussonnois force-pushed the KAFKA-4794 branch from 9f38088 to a2fdc01 Compare September 5, 2017 18:36

rhauch reviewed Sep 5, 2017

View reviewed changes

fhussonnois force-pushed the KAFKA-4794 branch from a2fdc01 to 10d7118 Compare September 5, 2017 20:42

rhauch reviewed Sep 5, 2017

View reviewed changes

Comment thread connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java Outdated

fhussonnois force-pushed the KAFKA-4794 branch from 10d7118 to cfc2b60 Compare September 5, 2017 21:10

ewencp added the connect label Feb 28, 2018

fhussonnois force-pushed the KAFKA-4794 branch from cfc2b60 to 417aa04 Compare January 18, 2019 00:11

fhussonnois force-pushed the KAFKA-4794 branch from 417aa04 to ee6da65 Compare March 11, 2020 19:46

fhussonnois force-pushed the KAFKA-4794 branch from ee6da65 to 4cf3d54 Compare March 22, 2020 10:48

rhauch reviewed May 7, 2020

View reviewed changes

rhauch approved these changes May 21, 2020

View reviewed changes

rhauch requested a review from kkonstantine May 21, 2020 23:16

kkonstantine reviewed May 24, 2020

View reviewed changes

fhussonnois and others added 5 commits May 24, 2020 11:34

KAFKA-4794: Add access to OffsetStorageReader from SourceConnector

601e6f6

Add two interfaces SinkConnectorContext/SourceConnectContext that extend ConnectorContext in order to expose an OffsetStorageReader instance.

KAFKA-4794: Avoid unnecessary lines

868d6fe

KAFKA-4794: Simplify SourceConnectorContext and SinkConnectorContext …

d1d6dee

…initialization in WorkerConnector

KAFKA-4794: Add unit tests for Connector, SinkConnector and SourceCon…

d8b4c26

…nector default methods

KAFKA-4794: Removed ambiguous case of connector not implementing sour…

7784286

…ce or sink, and incorporated other feedback

rhauch force-pushed the KAFKA-4794 branch from b7f45d2 to 7784286 Compare May 24, 2020 16:52

kkonstantine approved these changes May 24, 2020

View reviewed changes

rhauch merged commit 3818770 into apache:trunk May 25, 2020

kkonstantine mentioned this pull request May 25, 2020

KAFKA-5295: Allow source connectors to specify topic-specific settings for new topics (KIP-158) #8722

Merged

3 tasks

C0urante mentioned this pull request Mar 2, 2022

KAFKA-13510: Connect APIs to list all connector plugins and retrieve … #11572

Merged

3 tasks

-            final ConnectorContext delegateCtx = new DelegateToWorkerConnectorContext();
-            if (isSinkConnector()) {
-                SinkConnectorConfig.validate(config);
-                connector.initialize(new DelegateSinkConnectorContext(delegateCtx));
-            } else {
-                connector.initialize(new DelegateSourceConnectorContext(delegateCtx, offsetStorageReader));
-            }
+            if (isSinkConnector()) {
+                SinkConnectorConfig.validate(config);
+                connector.initialize(new DelegateSinkConnectorContext());
+            } else if (isSourceConnector()) {
+                connector.initialize(new DelegateSourceConnectorContext(offsetStorageReader));
+            } else {
+                connector.initialize(new DelegateToWorkerConnectorContext());
+            }

Conversation

fhussonnois commented Feb 27, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

asfbot commented Feb 27, 2017

Uh oh!

asfbot commented Feb 27, 2017

Uh oh!

asfbot commented Feb 27, 2017

Uh oh!

hleb-albau commented Aug 24, 2017

Uh oh!

fhussonnois commented Aug 31, 2017

Uh oh!

rhauch 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!

tedyu Sep 5, 2017

Choose a reason for hiding this comment

Uh oh!

fhussonnois commented Sep 5, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

fhussonnois commented Sep 7, 2017

Uh oh!

rhauch commented Sep 7, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

frankcoutinho commented Mar 5, 2020

Uh oh!

rhauch commented Mar 5, 2020

Uh oh!

rhauch commented May 6, 2020

Uh oh!

rhauch left a comment

Choose a reason for hiding this comment

Uh oh!

rhauch May 7, 2020

Choose a reason for hiding this comment

Uh oh!

rhauch May 7, 2020

Choose a reason for hiding this comment

Uh oh!

rhauch May 7, 2020

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

rhauch commented May 21, 2020

Uh oh!

rhauch commented May 21, 2020

Uh oh!

rhauch left a comment

Choose a reason for hiding this comment

Uh oh!

fhussonnois commented May 21, 2020

Uh oh!

rhauch commented May 23, 2020

Uh oh!

rhauch commented May 24, 2020

Uh oh!

kkonstantine left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

fhussonnois commented Feb 27, 2017 •

edited

Loading

fhussonnois commented Sep 5, 2017 •

edited

Loading

rhauch commented Sep 7, 2017 •

edited

Loading