KAFKA-9422: Track the set of topics a connector is using (KIP-558) by kkonstantine · Pull Request #8017 · apache/kafka

kkonstantine · 2020-01-29T03:29:11Z

This feature corresponds to KIP-558 and extends how the status.storage.topic is used to include information that allows Kafka Connect to keep track which topics a connector is using.

The set of topics a connector is actively using, is exposed via a new endpoint that is added to the REST API of Connect workers.

A GET /connectors/{name}/topics request will return the set of topics that have been recorded as active since a connector started or since the set of topics was reset for this connector.

An additional endpoints allows users to reset the set of active topics for a connector via the second endpoint that this feature is adding:

A PUT /connectors/{name}/topics/reset request clears the set of active topics.

An operator may enable or disable this feature by setting:
topic.tracking.enable (true by default).

Additionally, an operator may disable only reset requests via the Connect REST API by setting:
topic.tracking.allow.reset to false (true by default).

In the current commit the code for KIP-558 is feature complete.

It's tested by extending the current unit tests. Because changes are required in the integration testing framework itself in order to add integration testing for topic tracking (KIP-558), these tests will be submitted in a follow up pull request. Same, for Connect system tests that will extend coverage beyond integration testing. System tests will be added in a separate pull request, if additional coverage is required beyond integration testing.

Committer Checklist (excluded from commit message)

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

kkonstantine · 2020-01-29T03:41:48Z

cc @rhauch this is ready for a first pass. thx!

kkonstantine · 2020-01-29T19:07:24Z

retest this please

kkonstantine · 2020-01-29T19:53:26Z

note: That's a simple rename, because the member variable topic is easy to be masked by a local variable with this new feature.

ijuma · 2020-01-29T20:25:28Z

ok to test

…disabled

kkonstantine · 2020-01-29T21:33:23Z

ok to test

rhauch

Fantastic work, @kkonstantine. I like the simplicity of this implementation of KIP-558. I do have a few questions/suggestions below, all of which are pretty minor.

rhauch · 2020-01-29T22:47:01Z

+            expect.andAnswer(() -> new TopicStatus(
+                    topicCapture.getValue(),
+                    new ConnectorTaskId(connectorCapture.getValue(), 0),
+                    Time.SYSTEM.milliseconds()));


WDYT about doing something like the following to reduce the duplicate code:

Supplier<TopicStatus> returnStatus = () -> new TopicStatus( topicCapture.getValue(), new ConnectorTaskId(connectorCapture.getValue(), 0), Time.SYSTEM.milliseconds()); if (anyTimes) { expect.AndStubAnswer(returnStatus); } else { expect.AndAnswer(returnStatus); }

and doing the same in the other two similar methods?

I'm fuzzy on the details right now, but seems that it can't be inferred if it's not given inline.
Maybe we can optimize in a follow up that will include the other calls that add two implementations of IAnswer

Yeah, this was a minor change while leaving the fact there were still two other similar methods. We maybe should refactor later.

kkonstantine

Thanks @rhauch for the quick review.
Please find replies inline

kkonstantine · 2020-01-30T00:07:30Z

+            expect.andAnswer(() -> new TopicStatus(
+                    topicCapture.getValue(),
+                    new ConnectorTaskId(connectorCapture.getValue(), 0),
+                    Time.SYSTEM.milliseconds()));


I'm fuzzy on the details right now, but seems that it can't be inferred if it's not given inline.
Maybe we can optimize in a follow up that will include the other calls that add two implementations of IAnswer

kkonstantine · 2020-01-30T00:17:09Z

-                      RetryWithToleranceOperator retryWithToleranceOperator) {
+                      RetryWithToleranceOperator retryWithToleranceOperator,
+                      Time time,
+                      StatusBackingStore statusBackingStore) {


My main issue is that the new methods seem to belong to the StatusBackingStore interface. Fleshing them out to a different one (let's say TopicTracker) would would separate concerns here, but would also add a step of redirection that might be more complex to follow. wdyt?

kkonstantine · 2020-01-30T00:21:15Z

+            Map<String, Object> topicStatusMetadata = (Map<String, Object>) innerValue;
+            return new TopicStatus((String) topicStatusMetadata.get(TOPIC_NAME_KEY),
+                    (String) topicStatusMetadata.get(TOPIC_CONNECTOR_KEY),
+                    ((Long) topicStatusMetadata.get(TOPIC_TASK_KEY)).intValue(),


My intention was also to be more fault tolerant. But I also didn't want to diverge from the pattern used for the rest of the values here (e.g. #parseConnectorStatus). I'd suggest refactoring in a separate PR for all.

…sabled

rhauch

LGTM. Great work, @kkonstantine!

rhauch · 2020-01-30T01:44:25Z

ok to test

rhauch · 2020-01-30T01:49:17Z

retest this please

rhauch · 2020-01-30T05:45:02Z

The PR doesn't show the builds results, but:

multiple JDK 8 local builds passed in a row with no issues, but then one test failure in MirrorConnectorsIntegrationTest (see below)
Jenkins build with JDK 11 / Scala 2.13 passed
Jenkins build with JDK 8 / Scala 2.12 failed with one test failure in MirrorConnectorsIntegrationTest (see below).

The MirrorConnectorsIntegrationTest test failure appears to be a flaky test (see previously reported KAFKA-9013) that fails with:

java.lang.RuntimeException: Could not find enough records. found 0, expected 100
	at org.apache.kafka.connect.util.clusters.EmbeddedKafkaCluster.consume(EmbeddedKafkaCluster.java:334)
	at org.apache.kafka.connect.mirror.MirrorConnectorsIntegrationTest.testReplication(MirrorConnectorsIntegrationTest.java:198)

I tried 6 builds of the trunk branch, and 4 passed and 2 failed with the same apparently flaky test.

Therefore, I don't think this PR caused the flaky test, so I will merge.

Conflicts and/or compiler errors due to the fact that we temporarily reverted the commit that removes Scala 2.11 support: * SslAdminIntegrationTest: keep using JAdminClient, take upstream changes otherwise. * ReassignPartitionsClusterTest: keep using JAdminClient, take upstream changes otherwise. * KafkaApis: use `asScala.foreach` instead of `forEach`. # By Ismael Juma (3) and others # Via GitHub * apache-github/trunk: (22 commits) KAFKA-9437; Make the Kafka Protocol Friendlier with L7 Proxies [KIP-559] (apache#7994) KAFKA-9375: Add names to all Connect threads (apache#7901) MINOR: Introduce 2.5-IV0 IBP (apache#8010) KAFKA-8503; Add default api timeout to AdminClient (KIP-533) (apache#8011) Add retries to release.py script (apache#8021) KAFKA-8162: IBM JDK Class not found error when handling SASL (apache#6524) MINOR: Add explicit result type in public defs/vals (apache#7993) KAFKA-9408: Use StandardCharsets.UTF-8 instead of "UTF-8" (apache#7940) KAFKA-9474: Adds 'float64' to the RPC protocol types (apache#8012) KAFKA-9360: Allow disabling MM2 heartbeat and checkpoint emissions (apache#7887) KAFKA-7658: Add KStream#toTable to the Streams DSL (apache#7985) KAFKA-9445: Allow adding changes to allow serving from a specific partition (apache#7984) KAFKA-9422: Track the set of topics a connector is using (KIP-558) (apache#8017) KAFKA-9040; Add --all option to config command (apache#7607) KAFKA-4203: Align broker default for max.message.bytes with Java producer default (apache#4154) KAFKA-9426: Use switch instead of chained if/else in OffsetsForLeaderEpochClient (apache#7959) KAFKA-9405: Use Map.computeIfAbsent where applicable (apache#7937) KAFKA-9026: Use automatic RPC generation in DescribeAcls (apache#7560) MINOR: Remove unused fields in StreamsMetricsImpl (apache#7992) KAFKA-9460: Enable only TLSv1.2 by default and disable other TLS protocol versions (KIP-553) (apache#7998) ...

kkonstantine requested a review from rhauch January 29, 2020 19:08

kkonstantine commented Jan 29, 2020

View reviewed changes

kkonstantine added 3 commits January 29, 2020 15:46

KAFKA-9422: Track the set of topics a connector is using (KIP-558)

49dfb95

KAFKA-9422: Respond with error on topic/reset requests when reset is …

ac9e125

…disabled

KAFKA-9422: Fix formatting

228f634

kkonstantine force-pushed the kafka-9422 branch from a8a45ff to 228f634 Compare January 29, 2020 20:47

KAFKA-9422: Fix unchecked cast warning and unused imports

ec119b9

rhauch reviewed Jan 29, 2020

View reviewed changes

kkonstantine added 2 commits January 29, 2020 19:25

KAFKA-9422: Address comments

7c36f7b

KAFKA-9422: Add a test on topic status records

9529294

kkonstantine commented Jan 30, 2020

View reviewed changes

KAFKA-9422: Return error on topics endpoint when topic tracking is di…

7328b17

…sabled

rhauch approved these changes Jan 30, 2020

View reviewed changes

rhauch merged commit 7746301 into apache:trunk Jan 30, 2020

kkonstantine added the connect label Oct 16, 2020

Conversation

kkonstantine commented Jan 29, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

kkonstantine commented Jan 29, 2020

Uh oh!

kkonstantine commented Jan 29, 2020

Uh oh!

kkonstantine Jan 29, 2020

Choose a reason for hiding this comment

Uh oh!

ijuma commented Jan 29, 2020

Uh oh!

kkonstantine commented Jan 29, 2020

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!

rhauch Jan 29, 2020

Choose a reason for hiding this comment

Uh oh!

kkonstantine Jan 30, 2020

Choose a reason for hiding this comment

Uh oh!

rhauch Jan 30, 2020

Choose a reason for hiding this comment

Uh oh!

Uh oh!

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!

kkonstantine Jan 30, 2020

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

kkonstantine Jan 30, 2020

Choose a reason for hiding this comment

Uh oh!

kkonstantine Jan 30, 2020

Choose a reason for hiding this comment

Uh oh!

rhauch left a comment

Choose a reason for hiding this comment

Uh oh!

rhauch commented Jan 30, 2020

Uh oh!

rhauch commented Jan 30, 2020

Uh oh!

rhauch commented Jan 30, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

kkonstantine commented Jan 29, 2020 •

edited

Loading

rhauch commented Jan 30, 2020 •

edited

Loading