feat: error handling in sink connector by Emrehzl94 · Pull Request #142 · neo4j/neo4j-kafka-connector

Emrehzl94 · 2024-06-20T12:27:23Z

This pr includes error handling mechanism for sink connector and its integration tests. With this pr, now we are able to send error information to a configured DLQ for each sink message.

Co-authored-by: Emre Hizal <emre.hizal@neo4j.com> Co-authored-by: Florent Biville <florent.biville@neo4j.com>

ali-ince

Looks good. Left a couple of minor comments.

I think test names can be slightly rephrased in general.

ali-ince · 2024-06-24T11:34:11Z

sink/src/main/kotlin/org/neo4j/connectors/kafka/sink/strategy/CdcSchemaHandler.kt

  }

  override fun transformUpdate(event: NodeEvent): Query {
+    if (event.before == null) {


should we also have a similar validation for create and delete events, maybe in a follow-up PR. wdyt?

yes, I think we should, since they can also fail if before or after is null

ali-ince · 2024-06-24T11:46:39Z

sink-connector/src/test/kotlin/org/neo4j/connectors/kafka/sink/Neo4jSinkErrorIT.kt

+      errorDlqTopic = DLQ_TOPIC,
+      errorDlqContextHeadersEnable = true)
+  @Test
+  fun `should report errors when cypher strategy with multiple events`(


I think the test name requires a bit of rephrasing?

any suggestions? maybe should report failed events with cypher strategy ?

yep, that sounds better :)
bad news is that, most of the test cases in this class requires similar renaming :)

no worries :) not a big effort to change them :)

ali-ince · 2024-06-24T11:48:35Z

sink-connector/src/test/kotlin/org/neo4j/connectors/kafka/sink/Neo4jSinkErrorIT.kt

+          errorHeaders.getValue(ErrorHeaders.EXCEPTION_CLASS_NAME) shouldBe
+              "org.neo4j.driver.exceptions.ClientException"
+          errorHeaders.getValue(ErrorHeaders.EXCEPTION_MESSAGE) shouldBe
+              "Unable to convert kotlin.Unit to Neo4j Value."


Shall we have a better error message displayed here, wdyt?

so you mean we should prevent this error with some validation even before sending the query to the driver, if I'm not mistaken?

No, what I mean is that we should handle this error way before we send the query to Neo4j - it will still fail but with a clear error message about what's wrong.

got it 👍

ali-ince · 2024-06-24T11:49:53Z

sink-connector/src/test/kotlin/org/neo4j/connectors/kafka/sink/Neo4jSinkErrorIT.kt

+          errorHeaders.getValue(ErrorHeaders.EXCEPTION_CLASS_NAME) shouldBe
+              "org.neo4j.driver.exceptions.ClientException"
+          errorHeaders.getValue(ErrorHeaders.EXCEPTION_MESSAGE) shouldBe
+              "Unable to convert kotlin.Unit to Neo4j Value."


ali-ince · 2024-06-24T11:54:03Z

sink-connector/src/test/kotlin/org/neo4j/connectors/kafka/sink/Neo4jSinkErrorIT.kt

+    TopicVerifier.create<String, String>(consumer)
+        .assertMessageValue { it shouldBe message2ToFail.value }
+        .verifyWithin(Duration.ofSeconds(30))
+  }


maybe for a future improvement but we could also verify that the connector instance is in FAILED state.

should I create a card for this?

neo4j-connectors · 2024-06-26T18:39:56Z

	Warnings
⚠️	Commit Message 'refactor: implement requested changes.': subject may not end with full stop

Generated by 🚫 dangerJS against 65300e8

# Conflicts: # sink/src/main/kotlin/org/neo4j/connectors/kafka/sink/strategy/RedirectingHandler.kt

ali-ince · 2024-06-27T09:55:47Z

sink/src/test/kotlin/org/neo4j/connectors/kafka/sink/strategy/CdcSourceIdHandlerTest.kt

  @Test
  fun `should split changes into transactional boundaries`() {
-    val handler = CdcSchemaHandler("my-topic", Renderer.getRenderer(Configuration.defaultConfig()))
+    val handler =


nice spot 😉

ali-ince

Looks good to me. Let's wait for @fbiville's approval before merge though.

fbiville · 2024-06-27T11:31:41Z

sink-connector/src/test/kotlin/org/neo4j/connectors/kafka/sink/Neo4jSinkErrorIT.kt

+  @Test
+  fun `should report an error with all error headers when headers are enabled`(
+      @TopicProducer(TOPIC) producer: ConvertingKafkaProducer,
+      @TopicConsumer(topic = DLQ_TOPIC, offset = "earliest") consumer: ConvertingKafkaConsumer,


Suggested change

@TopicConsumer(topic = DLQ_TOPIC, offset = "earliest") consumer: ConvertingKafkaConsumer,

@TopicConsumer(topic = DLQ_TOPIC, offset = "earliest") errorConsumer: ConvertingKafkaConsumer,

fbiville · 2024-06-27T11:32:36Z

sink-connector/src/test/kotlin/org/neo4j/connectors/kafka/sink/Neo4jSinkErrorIT.kt

+  ) {
+    session.run("CREATE CONSTRAINT FOR (n:Person) REQUIRE n.id IS KEY").consume()
+
+    val schema =


Suggested change

val schema =

val schemaWithMissingSurname =

fbiville · 2024-06-27T11:33:08Z

sink-connector/src/test/kotlin/org/neo4j/connectors/kafka/sink/Neo4jSinkErrorIT.kt

+              "org.neo4j.driver.exceptions.ClientException"
+          errorHeaders.getValue(ErrorHeaders.EXCEPTION_MESSAGE) shouldBe
+              """Cannot merge the following node because of null property value for 'surname': (:Person {surname: null})"""
+          errorHeaders.getValue(ErrorHeaders.EXCEPTION_STACKTRACE) shouldNotBe null


why null?

you mean why we only check it's not null?

I'm wondering why we don't get a stack trace

we're getting it, just it's gonna be too many lines in the test to put the all stack trace here and check they are equal, this is why I didn't put it and only check it's not null, maybe instead of all stack trace I can put shouldContain check

sink-connector/src/test/kotlin/org/neo4j/connectors/kafka/sink/Neo4jSinkErrorIT.kt

fbiville · 2024-06-27T11:37:48Z

sink-connector/src/test/kotlin/org/neo4j/connectors/kafka/sink/Neo4jSinkErrorIT.kt

+  }
+
+  @Neo4jSink(
+      cud = [CudStrategy(TOPIC)], errorDlqTopic = DLQ_TOPIC, errorDlqContextHeadersEnable = true)


maybe it's just me but I have a hard time parsing errorDlqContextHeadersEnable 😅

Any suggestions for renaming? Yes, I know it's hard to grab at first glance but I just followed same pattern with actual configuration name errors.deadletterqueue.context.headers.enable 😄

based on my quick reading of https://www.confluent.io/blog/kafka-connect-deep-dive-error-handling-dead-letter-queues/#92942d30-20e8-4c52-80f5-924b639e639b, something like includeErrorMessage or sth?

what about enableErrorHeaders ?

fbiville · 2024-06-27T11:42:12Z

sink/src/test/kotlin/org/neo4j/connectors/kafka/sink/strategy/CdcSchemaHandlerTest.kt

+      handler.handle(listOf(nodeChangeEventMessage))
+    } shouldHaveMessage "create operation requires 'after' field in the event object"
+
+    val relationshipChangeEventMessage =


I'd personally move this to a separate test

fbiville · 2024-06-27T11:42:24Z

sink/src/test/kotlin/org/neo4j/connectors/kafka/sink/strategy/CdcSchemaHandlerTest.kt

+      handler.handle(listOf(nodeChangeEventMessage))
+    } shouldHaveMessage "update operation requires 'after' field in the event object"
+
+    val relationshipChangeEventMessage =


same comment as above

fbiville · 2024-06-27T11:42:58Z

sink/src/test/kotlin/org/neo4j/connectors/kafka/sink/strategy/CdcSourceIdHandlerTest.kt

+      handler.handle(listOf(nodeChangeEventMessage))
+    } shouldHaveMessage "create operation requires 'after' field in the event object"
+
+    val relationshipChangeEventMessage =


same comment as above

fbiville · 2024-06-27T11:43:05Z

sink/src/test/kotlin/org/neo4j/connectors/kafka/sink/strategy/CdcSourceIdHandlerTest.kt

+      handler.handle(listOf(nodeChangeEventMessage))
+    } shouldHaveMessage "update operation requires 'before' field in the event object"
+
+    val relationshipChangeEventMessage =


fbiville · 2024-06-27T11:43:11Z

sink/src/test/kotlin/org/neo4j/connectors/kafka/sink/strategy/CdcSourceIdHandlerTest.kt

+      handler.handle(listOf(nodeChangeEventMessage))
+    } shouldHaveMessage "update operation requires 'after' field in the event object"
+
+    val relationshipChangeEventMessage =


ali-ince

Looks good, I think this is now ready to merged 👍

ali-ince and others added 9 commits June 20, 2024 14:02

refactor: rework cud handling

d4a6105

style: add license headers

ac2844e

refactor: use type casting instead of deserialization

b602638

feat: relate change queries to messages

6c314e1

Co-authored-by: Emre Hizal <emre.hizal@neo4j.com> Co-authored-by: Florent Biville <florent.biville@neo4j.com>

test: set up test environment for error testing

8fdcdaf

test: start to add test cases for error handling

5ca3cc5

test: add error headers and more test cases

ed332d7

refactor: add data classes to pass down the event message

286f095

test: add more test cases for error handling

98c246d

Emrehzl94 self-assigned this Jun 20, 2024

Emrehzl94 marked this pull request as ready for review June 20, 2024 12:37

Emrehzl94 requested a review from a team as a code owner June 20, 2024 12:37

test: fix sink registration test

0fb4321

ali-ince requested changes Jun 24, 2024

View reviewed changes

refactor: implement requested changes.

6e8cea6

Emrehzl94 added 3 commits June 26, 2024 20:48

Merge branch 'main' into error-handling

113b92d

# Conflicts: # sink/src/main/kotlin/org/neo4j/connectors/kafka/sink/strategy/RedirectingHandler.kt

fix: remove unnecessary handler

b9b9ffc

test: fix failing tests

0984c7b

ali-ince reviewed Jun 27, 2024

View reviewed changes

ali-ince approved these changes Jun 27, 2024

View reviewed changes

fbiville reviewed Jun 27, 2024

View reviewed changes

refactor: implement requested changes

65300e8

ali-ince approved these changes Jun 28, 2024

View reviewed changes

Emrehzl94 merged commit d6df239 into neo4j:main Jun 28, 2024

ali-ince deleted the error-handling branch July 2, 2024 11:33

	@TopicConsumer(topic = DLQ_TOPIC, offset = "earliest") consumer: ConvertingKafkaConsumer,
	@TopicConsumer(topic = DLQ_TOPIC, offset = "earliest") errorConsumer: ConvertingKafkaConsumer,

Conversation

Emrehzl94 commented Jun 20, 2024

Uh oh!

ali-ince 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!

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

neo4j-connectors commented Jun 26, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ali-ince 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!

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!

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!

Choose a reason for hiding this comment

Uh oh!

ali-ince left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

neo4j-connectors commented Jun 26, 2024 •

edited

Loading