KAFKA-9971: Error Reporting in Sink Connectors (KIP-610) by aakashnshah · Pull Request #8720 · apache/kafka

aakashnshah · 2020-05-23T01:00:41Z

Implementation for KIP-610: https://cwiki.apache.org/confluence/display/KAFKA/KIP-610%3A+Error+Reporting+in+Sink+Connectors

This PR adds the ErrantRecordReporter interface as well as its implementation - WorkerErrantRecordReporter. The WorkerErrantRecordReporter is created in Worker and brought up through WorkerSinkTask to WorkerSinkTaskContext. An integration test and unit test has been added.

levzem

@aakashnshah looks great! excited for this feature

just got mostly nits for you

levzem · 2020-05-25T00:31:17Z

extract to variable

levzem · 2020-05-25T00:32:23Z

do we need this invalid config step here

I don't think it hurts to add this step.

rhauch

Thanks, @aakashnshah! I have a few comments below, including a kind of big question: why are we not reusing the existing RetryWithToleranceOperator that already tracks the tolerance and 1 or more reporters?

rhauch · 2020-05-25T15:02:14Z

All fields that can be final should be marked as such. This provides semantic intent to future developers and helps prevent unintentionally changing the fields in the future.

rhauch · 2020-05-25T15:05:44Z

Do we really want to pass the ExecutionException to the ConnectException, or would it be better to pass that exception's cause to the ConnectException?

How about log messages here?

rhauch · 2020-05-25T15:24:35Z

I suggested earlier about reusing the RetryWithToleranceOperator, and that doing so might require adding a produce-like method to that class that simply reports a new error. If that method took a Callback here and passed it to its producer.send(...) call, then we could provide a callback that removed the (completed) future from our list, helping to keep that list as small as possible with only the incomplete futures.

If we did that, we'd want to use a LinkedList rather than an ArrayList, since we're no longer removing futures only from the ends of the list.

rhauch · 2020-05-25T15:25:53Z

If we ensure that the producer future is never null, then we can remove the if (future == null) kind of checks in this class' methods.

gharris1727

Please add your new API class to the isolation whitelist, and update the corresponding tests.
https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginUtils.java#L130-L143

gharris1727 · 2020-05-26T23:23:22Z

Does this test ever encounter this exception? I don't think we will be able to backport this test to < 2.6 because the method won't exist at all, much less generate the exception that is being caught here.
If anything, this generates a less informative NPE later in put, and hides the actual root cause.

Yeah, I should remove this comment. This test won't encounter this exception since it's always going to have the class and method if the test itself exists.

@aakashnshah let's remove this comment

gharris1727 · 2020-05-26T23:27:35Z

Does this have an unbounded waiting time? How does this interact with task.shutdown.graceful.timeout.ms? What is the delivery guarantee of these error reports?

Yes, the waiting time is unbounded. The delivery guarantee is that all errant records up to the latest offset in preCommit() will be sent to Kafka before preCommit() is invoked.

rhauch · 2020-05-27T15:01:59Z

ok to test

rhauch

Thanks, @aakashnshah. I think this is much closer to a more straightforward implementation of the KIP.

I took another pass, and there are quite a few things to change/correct. I'll keep reviewing the test code in a subsequent review.

rhauch · 2020-05-27T15:06:23Z

You can't change this public API. InternalSinkRecord needs to be in the runtime module.

rhauch · 2020-05-27T15:06:52Z

When you move InternalSinkRecord to the runtime module, be sure to make this private final.

rhauch

Made another pass, with comments inline below. Thanks, @aakashnshah.

rhauch · 2020-05-27T18:52:40Z

Let's add a trace log message before and after this call.

rhauch · 2020-05-27T19:07:52Z

@aakashnshah let's remove this comment

Signed-off-by: Aakash Shah <ashah@confluent.io>

rhauch

A few more suggestions

rhauch · 2020-05-27T19:59:56Z

retest this please

rhauch · 2020-05-27T20:00:08Z

ok to test

rhauch · 2020-05-27T20:00:44Z

retest this please

rhauch · 2020-05-27T20:04:11Z

ok to test

rhauch · 2020-05-27T20:04:25Z

retest this please

rhauch · 2020-05-27T20:18:38Z

retest this please

rhauch · 2020-05-27T21:32:59Z

Retest this please.

rhauch · 2020-05-27T21:36:50Z

ok to test

rhauch · 2020-05-27T21:40:04Z

Closing to try to fix builds; will reopen shortly.

rhauch · 2020-05-27T21:41:11Z

ok to test

rhauch · 2020-05-27T21:42:16Z

retest this please

rhauch · 2020-05-27T22:11:43Z

retest this please.

rhauch · 2020-05-27T22:11:58Z

retest this please

rhauch

A few more fixes. It's getting close.

rhauch · 2020-05-28T00:19:04Z

Not sure what's happening, but here are some finally-running builds from previous commit (0e40):

kkonstantine

Thanks for the PR @aakashnshah
Looks good overall.
I left a few mostly minor comments after a quick pass, if there's time to address.

kkonstantine · 2020-05-28T00:17:44Z

+    @Override
+    public boolean equals(Object o) {
+        return super.equals(o);
+    }
+
+    @Override
+    public int hashCode() {
+        return super.hashCode();
+    }
+
+    @Override
+    public String toString() {
+        return super.toString();
+    }


these overrides don't seem to add much.

IIUC, spotbugs complained if these were not here.

never mind then. I'll leave this to AI.

kkonstantine · 2020-05-28T00:29:32Z

Also, topic can never be null if it's coming from a parsed config value that doesn't have null as its default value. (another way to think of that is that you can't pass a null value from properties)

kkonstantine · 2020-05-28T02:03:22Z

+
+        // delete connector
+        connect.deleteConnector(CONNECTOR_NAME);
+


Suggested change

rhauch · 2020-05-28T03:18:17Z

Three green builds of commit 126b04:

@aakashnshah, let's get fixes for @kkonstantine's suggestions. Thanks!

rhauch · 2020-05-28T04:07:35Z

ok to test

rhauch · 2020-05-28T04:08:04Z

retest this please

rhauch · 2020-05-28T04:11:29Z

retest this please

rhauch

Thanks, @aakashnshah! LGTM, pending green builds (that are finally running!)

kkonstantine

LGTM
Nicely done @aakashnshah

kkonstantine · 2020-05-28T06:42:22Z

2/3 builds are green. Merging now to allow for other PRs to resolve conflicts sooner rather than later, if needed.

* 'trunk' of github.com:apache/kafka: (36 commits) Remove redundant `containsKey` call in KafkaProducer (apache#8761) KAFKA-9494; Include additional metadata information in DescribeConfig response (KIP-569) (apache#8723) KAFKA-10061; Fix flaky `ReassignPartitionsIntegrationTest.testCancellation` (apache#8749) KAFKA-9130; KIP-518 Allow listing consumer groups per state (apache#8238) KAFKA-9501: convert between active and standby without closing stores (apache#8248) KAFKA-10056; Ensure consumer metadata contains new topics on subscription change (apache#8739) MINOR: Log the reason for coordinator discovery failure (apache#8747) KAFKA-10029; Don't update completedReceives when channels are closed to avoid ConcurrentModificationException (apache#8705) MINOR: remove unnecessary timeout for admin request (apache#8738) MINOR: Relax Percentiles test (apache#8748) MINOR: regression test for task assignor config (apache#8743) MINOR: Update documentation.html to refer to 2.6 (apache#8745) MINOR: Update documentation.html to refer to 2.5 (apache#8744) KAFKA-9673: Filter and Conditional SMTs (apache#8699) KAFKA-9971: Error Reporting in Sink Connectors (KIP-610) (apache#8720) KAFKA-10052: Harden assertion of topic settings in Connect integration tests (apache#8735) MINOR: Slight MetadataCache tweaks to avoid unnecessary work (apache#8728) KAFKA-9802; Increase transaction timeout in system tests to reduce flakiness (apache#8736) KAFKA-10050: kafka_log4j_appender.py fixed for JDK11 (apache#8731) KAFKA-9146: Add option to force delete active members in StreamsResetter (apache#8589) ... # Conflicts: # core/src/main/scala/kafka/log/Log.scala

kkonstantine added the connect label May 24, 2020

levzem reviewed May 25, 2020

View reviewed changes

rhauch reviewed May 25, 2020

View reviewed changes

gharris1727 reviewed May 26, 2020

View reviewed changes

gharris1727 reviewed May 27, 2020

View reviewed changes

aakashnshah force-pushed the dlq branch from 4c51b5e to 622bc46 Compare May 27, 2020 06:07

rhauch reviewed May 27, 2020

View reviewed changes

aakashnshah force-pushed the dlq branch from 622bc46 to a0578a0 Compare May 27, 2020 18:26

rhauch reviewed May 27, 2020

View reviewed changes

aakashnshah added 4 commits May 27, 2020 12:26

KAFKA-9971: Error Reporting in Sink Connectors

2603450

Signed-off-by: Aakash Shah <ashah@confluent.io>

addressed comments

1134e99

Signed-off-by: Aakash Shah <ashah@confluent.io>

addressed more comments

cd04316

Signed-off-by: Aakash Shah <ashah@confluent.io>

addressed some more comments

b91a8f9

aakashnshah force-pushed the dlq branch from a0578a0 to b91a8f9 Compare May 27, 2020 19:26

rhauch reviewed May 27, 2020

View reviewed changes

KAFKA-9971: Added trace logging

ab1af2b

more comments

0e40408

rhauch closed this May 27, 2020

rhauch reopened this May 27, 2020

rhauch reviewed May 27, 2020

View reviewed changes

more comments again

126b04c

kkonstantine reviewed May 28, 2020

View reviewed changes

addressed some more comments

1c4442f

rhauch approved these changes May 28, 2020

View reviewed changes

kkonstantine approved these changes May 28, 2020

View reviewed changes

kkonstantine merged commit 38c1e96 into apache:trunk May 28, 2020

kkonstantine changed the title ~~KAFKA-9971: Error Reporting in Sink Connectors~~ KAFKA-9971: Error Reporting in Sink Connectors (KIP-610) May 28, 2020

aakashnshah deleted the dlq branch June 7, 2020 22:51

chia7712 mentioned this pull request Oct 29, 2020

KAFKA-10658 ErrantRecordReporter.report always return completed futur… #9525

Merged

3 tasks

Conversation

aakashnshah commented May 23, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

levzem left a comment

Choose a reason for hiding this comment

Uh oh!

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!

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!

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!

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

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 May 27, 2020

Uh oh!

rhauch left a comment

Choose a reason for hiding this comment

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!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

rhauch left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

aakashnshah commented May 23, 2020 •

edited

Loading