From 63762e844e2254448f3325500de7d88b8e019aae Mon Sep 17 00:00:00 2001 From: Gunnar Morling Date: Tue, 11 Jun 2019 12:22:38 +0200 Subject: [PATCH 1/2] KAFKA-8523 Avoiding raw type usage --- .../kafka/connect/transforms/InsertFieldTest.java | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertFieldTest.java b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertFieldTest.java index a0a09752a4ae9..8f20a3783fa27 100644 --- a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertFieldTest.java +++ b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertFieldTest.java @@ -104,11 +104,10 @@ public void schemalessInsertConfiguredFields() { final SourceRecord transformedRecord = xform.apply(record); - assertEquals(42L, ((Map) transformedRecord.value()).get("magic")); - assertEquals("test", ((Map) transformedRecord.value()).get("topic_field")); - assertEquals(0, ((Map) transformedRecord.value()).get("partition_field")); - assertEquals(null, ((Map) transformedRecord.value()).get("timestamp_field")); - assertEquals("my-instance-id", ((Map) transformedRecord.value()).get("instance_id")); + assertEquals(42L, ((Map) transformedRecord.value()).get("magic")); + assertEquals("test", ((Map) transformedRecord.value()).get("topic_field")); + assertEquals(0, ((Map) transformedRecord.value()).get("partition_field")); + assertEquals(null, ((Map) transformedRecord.value()).get("timestamp_field")); + assertEquals("my-instance-id", ((Map) transformedRecord.value()).get("instance_id")); } - } From c327e315e58a15e129639dfaebcd25ca8b995656 Mon Sep 17 00:00:00 2001 From: Gunnar Morling Date: Fri, 27 Sep 2019 22:00:46 +0200 Subject: [PATCH 2/2] KAFKA-8523 Gracefully handling tombstone events in InsertField SMT --- .../kafka/connect/transforms/InsertField.java | 8 +++- .../connect/transforms/InsertFieldTest.java | 43 +++++++++++++++++++ 2 files changed, 50 insertions(+), 1 deletion(-) diff --git a/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/InsertField.java b/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/InsertField.java index 5e472a907c3f4..93ba79c0e3545 100644 --- a/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/InsertField.java +++ b/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/InsertField.java @@ -127,13 +127,19 @@ public void configure(Map props) { @Override public R apply(R record) { - if (operatingSchema(record) == null) { + if (isTombstoneRecord(record)) { + return record; + } else if (operatingSchema(record) == null) { return applySchemaless(record); } else { return applyWithSchema(record); } } + private boolean isTombstoneRecord(R record) { + return record.value() == null; + } + private R applySchemaless(R record) { final Map value = requireMap(operatingValue(record), PURPOSE); diff --git a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertFieldTest.java b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertFieldTest.java index 8f20a3783fa27..b22872cacb236 100644 --- a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertFieldTest.java +++ b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertFieldTest.java @@ -110,4 +110,47 @@ public void schemalessInsertConfiguredFields() { assertEquals(null, ((Map) transformedRecord.value()).get("timestamp_field")); assertEquals("my-instance-id", ((Map) transformedRecord.value()).get("instance_id")); } + + + @Test + public void insertConfiguredFieldsIntoTombstoneEventWithoutSchemaLeavesValueUnchanged() { + final Map props = new HashMap<>(); + props.put("topic.field", "topic_field!"); + props.put("partition.field", "partition_field"); + props.put("timestamp.field", "timestamp_field?"); + props.put("static.field", "instance_id"); + props.put("static.value", "my-instance-id"); + + xform.configure(props); + + final SourceRecord record = new SourceRecord(null, null, "test", 0, + null, null); + + final SourceRecord transformedRecord = xform.apply(record); + + assertEquals(null, transformedRecord.value()); + assertEquals(null, transformedRecord.valueSchema()); + } + + @Test + public void insertConfiguredFieldsIntoTombstoneEventWithSchemaLeavesValueUnchanged() { + final Map props = new HashMap<>(); + props.put("topic.field", "topic_field!"); + props.put("partition.field", "partition_field"); + props.put("timestamp.field", "timestamp_field?"); + props.put("static.field", "instance_id"); + props.put("static.value", "my-instance-id"); + + xform.configure(props); + + final Schema simpleStructSchema = SchemaBuilder.struct().name("name").version(1).doc("doc").field("magic", Schema.OPTIONAL_INT64_SCHEMA).build(); + + final SourceRecord record = new SourceRecord(null, null, "test", 0, + simpleStructSchema, null); + + final SourceRecord transformedRecord = xform.apply(record); + + assertEquals(null, transformedRecord.value()); + assertEquals(simpleStructSchema, transformedRecord.valueSchema()); + } }