Skip to content

feat: make CDC message key serialization configurable #52

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
fbiville merged 13 commits into from
Jan 22, 2024
Merged

feat: make CDC message key serialization configurable #52

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
9a156e1
feat: make CDC message key serialization configurable
fbiville Jan 5, 2024
1916394
test: add tests for configuration & implemented serializers
fbiville Jan 9, 2024
46ef718
feat: add unit tests and support element ID strategy
fbiville Jan 10, 2024
c3ae969
feat: support entity keys strategy
fbiville Jan 11, 2024
e23f492
test: add tests for entity key strategy
fbiville Jan 12, 2024
d42f8c1
chore: review feedback, rename strategy class
fbiville Jan 12, 2024
d049826
chore: undo IDE-generated formatting changes
fbiville Jan 12, 2024
9c7eab6
refactor: hardcode key deserializer as before
fbiville Jan 12, 2024
2b77304
refactor: create ChangeEvent in test data
fbiville Jan 12, 2024
31e7d9d
feat: nullify change keys for keyless entities
fbiville Jan 14, 2024
ca6c521
refactor: may the fold be with you
fbiville Jan 14, 2024
c5a550d
fix: review feedback - rename variable
fbiville Jan 19, 2024
54fdde0
fix: fix now-forbidden warnings 🙈
fbiville Jan 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 14 additions & 2 deletions common/src/main/kotlin/org/neo4j/connectors/kafka/data/ChangeEventExtensions.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,11 +125,17 @@ object ChangeEventExtensions {

private fun NodeEvent.toConnectValue(schema: Schema): Struct =
Struct(schema).also {
val keys =
if (this.keys.isEmpty()) {
null
} else {
DynamicTypes.valueFor(schema.field("keys").schema(), this.keys)
}
it.put("elementId", this.elementId)
it.put("eventType", this.eventType.name)
it.put("operation", this.operation.name)
it.put("labels", this.labels)
it.put("keys", DynamicTypes.valueFor(schema.field("keys").schema(), this.keys))
it.put("keys", keys)
it.put("state", nodeStateValue(schema.field("state").schema(), this.before, this.after))
}

Expand Down Expand Up @@ -208,13 +214,19 @@ object ChangeEventExtensions {

private fun RelationshipEvent.toConnectValue(schema: Schema): Struct =
Struct(schema).also {
val keys =
if (this.keys.isEmpty()) {
null
} else {
DynamicTypes.valueFor(schema.field("keys").schema(), this.keys)
}
it.put("elementId", this.elementId)
it.put("eventType", this.eventType.name)
it.put("operation", this.operation.name)
it.put("type", this.type)
it.put("start", this.start.toConnectValue(schema.field("start").schema()))
it.put("end", this.end.toConnectValue(schema.field("end").schema()))
it.put("keys", DynamicTypes.valueFor(schema.field("keys").schema(), this.keys))
it.put("keys", keys)
it.put(
"state",
relationshipStateValue(schema.field("state").schema(), this.before, this.after))
Expand Down
51 changes: 48 additions & 3 deletions common/src/test/kotlin/org/neo4j/connectors/kafka/data/ChangeEventExtensionsTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -929,6 +929,43 @@ class ChangeEventExtensionsTest {
.put("since", "2000-01-01"))))
}

@Test
fun `node event keys should be nullified when node keys are not defined`() {
val (_, _, schema, value) =
newChangeEvent(
NodeEvent(
"element-0",
EntityOperation.CREATE,
listOf("Label1", "Label2"),
mapOf(),
null,
NodeState(listOf("Label1"), mapOf("id" to 5L))))

val expectedKeySchema = SchemaBuilder.struct().optional().build()
schema.nestedSchema("event.keys") shouldBe expectedKeySchema
value.nestedValue("event.keys") shouldBe null
}

@Test
fun `relationship event keys should be nullified when rel keys are not defined`() {
val (_, _, schema, value) =
newChangeEvent(
RelationshipEvent(
"rel-0",
"WORKS_FOR",
Node("node-0", listOf("Person"), mapOf()),
Node("node-1", listOf("Company"), mapOf()),
listOf(),
EntityOperation.DELETE,
RelationshipState(mapOf("id" to 5L)),
null))

val expectedKeySchema =
SchemaBuilder.array(SchemaBuilder.struct().optional().build()).optional().build()
schema.nestedSchema("event.keys") shouldBe expectedKeySchema
value.nestedValue("event.keys") shouldBe null
}

data class ChangeEventResult<T : Event>(
val event: T,
val change: ChangeEvent,
Expand Down Expand Up @@ -964,12 +1001,20 @@ class ChangeEventExtensionsTest {
private fun Schema.nestedSchema(path: String): Schema {
require(path.isNotBlank())

var current = this
path.split('.').forEach { current = current.field(it).schema() }
return current
return path.split('.').fold(this) { schema, field -> schema.field(field).schema() }
Copy link
Contributor

@ali-ince ali-ince Jan 18, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👌

}

private fun Schema.nestedValueSchema(path: String): Schema {
return nestedSchema(path).valueSchema()
}

private fun Struct.nestedValue(path: String): Any? {
require(path.isNotBlank())

val fields = path.split('.')
return fields
.dropLast(1)
.fold(this) { struct, field -> struct[field] as Struct }
.get(fields.last())
}
}
1 change: 1 addition & 0 deletions legacy-connectors/src/test/kotlin/streams/kafka/connect/source/LegacyNeo4jSourceIT.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@ class LegacyNeo4jSourceIT {
mapOf("execId" to executionId))
.consume()

@Suppress("DEPRECATION")
TopicVerifier.create(consumer)
.expectMessageValueMatching { value ->
value.asMap().excludingKeys("timestamp") ==
Expand Down
Loading