fix: implement processMap function to MAP structured data #99

peacecwz · 2023-05-10T09:39:01Z

Maybe you saw If you are using HeaderToField as transforms and your data struct is Map, the plugin is throwing error like "MAP is unsupported ..." I implemented processMap function and I tried to use it and It works well. Maybe you want to merge it as contribution

jcustenborder · 2023-05-10T14:23:04Z

@peacecwz Happy to merge this! Would you mind adding some unit tests?

peacecwz · 2023-05-10T14:24:55Z

@jcustenborder Actually no but I can add some unit tests for the function. I'll update the PR quickly

peacecwz · 2023-05-10T14:44:45Z

@jcustenborder I added test for processMap Is that okay for that?

peacecwz · 2023-05-10T15:13:20Z

@jcustenborder btw If this PR will be merged can you also make a release it? Because I'm using my private artifact. I couldn't run Jenkins pipeline. It uploaded the artifact manually to S3 and deploy it as well but I would like to deploy with confluent-kafka CLI as officially

peacecwz · 2023-05-25T12:12:26Z

@jcustenborder Hey can you check the PR?

harpaj · 2023-09-13T09:32:21Z

src/main/java/com/github/jcustenborder/kafka/connect/transform/common/HeaderToField.java

+      });
+    }
+
+    input.put("_headers", headers);


I just tested this code because we are running into a similar problem.
It appears to add the extracted values to the new struct field _headers which you are setting here, creating a nested structure.
So

".header.mappings": "time:INT64:d,time:INT64:h,time:INT64:m"

becomes

"value": { "_headers" : { "d" : 1669852804800000000, "h" : 1669852804800000000, "m" : 1669852804800000000 } }

I would have expected a flat structure here (and interestingly also the test you added shows a flat structure).

The test actually doesn't test this correctly.
To have the same behaviour as for Structs (i,e, adding the new fields to the root), this should be:

Suggested change

input.put("_headers", headers);

input.putAll(headers);

okayhooni · 2023-10-20T11:02:36Z

@peacecwz

Could you fix same issue on the ChangeCase SMT..?

(If not.. I will try to implement processMap method on the ChangeCase, by referring to your commit.)

Caused by: java.lang.UnsupportedOperationException: MAP is not a supported type.
	at com.github.jcustenborder.kafka.connect.transform.common.BaseTransformation.processMap(BaseTransformation.java:39)
	at com.github.jcustenborder.kafka.connect.transform.common.BaseTransformation.process(BaseTransformation.java:120)
	at com.github.jcustenborder.kafka.connect.transform.common.ChangeCase$Value.apply(ChangeCase.java:128)
	at org.apache.kafka.connect.runtime.TransformationChain.lambda$apply$0(TransformationChain.java:50)
	at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:180)
	at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:214)
	... 15 more

greyfairer

See my suggestions to fix the test, because it doesn't test anything now.

And I'd also prefer the transformation to behave similar with Maps (JSON) as with Structs (AVRO), so add the field to the root of the object.

greyfairer · 2024-04-05T11:52:21Z

src/main/java/com/github/jcustenborder/kafka/connect/transform/common/HeaderToField.java

+      });
+    }
+
+    input.put("_headers", headers);


The test actually doesn't test this correctly.
To have the same behaviour as for Structs (i,e, adding the new fields to the root), this should be:

Suggested change

input.put("_headers", headers);

input.putAll(headers);

greyfairer · 2024-04-05T11:54:06Z

src/test/java/com/github/jcustenborder/kafka/connect/transform/common/HeaderToFieldTest.java

@@ -71,4 +68,44 @@ public void apply() throws IOException {
    assertStruct(expectedStruct, (Struct) actualRecord.value());
  }

+  @Test
+  public void applyWithMap() throws IOException {
+    this.transformation = new HeaderToField.Key<>();


Suggested change

this.transformation = new HeaderToField.Key<>();

this.transformation = new HeaderToField.Value<>();

greyfairer · 2024-04-05T11:55:36Z

src/test/java/com/github/jcustenborder/kafka/connect/transform/common/HeaderToFieldTest.java

+    ConnectHeaders inputHeaders = new ConnectHeaders();
+    inputHeaders.addString("applicationId", "testing");
+
+    Schema inputSchema = SchemaBuilder.map(SchemaBuilder.STRING_SCHEMA, SchemaBuilder.OPTIONAL_STRING_SCHEMA)


Suggested change

Schema inputSchema = SchemaBuilder.map(SchemaBuilder.STRING_SCHEMA, SchemaBuilder.OPTIONAL_STRING_SCHEMA)

Map<String, Object> inputSchema = new HashMap<>();

value.put("firstName", "example");

value.put("lastName", "user");

greyfairer · 2024-04-05T11:56:23Z

src/test/java/com/github/jcustenborder/kafka/connect/transform/common/HeaderToFieldTest.java

+
+    SinkRecord actualRecord = this.transformation.apply(inputRecord);
+    assertNotNull(actualRecord, "record should not be null.");
+    assertEquals(expectedSchema.parameters().size(), 3);


Suggested change

assertEquals(expectedSchema.parameters().size(), 3);

assertEquals("testing", ((Map<String, String>)actualRecord.value()).get("applicationId"));

greyfairer · 2024-04-10T14:30:50Z

@peacecwz see peacecwz#1

peacecwz added 2 commits May 10, 2023 11:30

fix: implement processMap function to MAP structured data

517ba74

fix: add header by field name

b788921

feat: add tests to HeaderToField for processMap function implementation

df4cbd0

harpaj reviewed Sep 13, 2023

View reviewed changes

jx2lee mentioned this pull request Oct 12, 2023

MAP support for HeaderToField #71

Open

greyfairer reviewed Apr 5, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix: implement processMap function to MAP structured data #99

fix: implement processMap function to MAP structured data #99

peacecwz commented May 10, 2023

jcustenborder commented May 10, 2023

peacecwz commented May 10, 2023

peacecwz commented May 10, 2023

peacecwz commented May 10, 2023

peacecwz commented May 25, 2023

harpaj Sep 13, 2023 •

edited

Loading

greyfairer Apr 5, 2024

okayhooni commented Oct 20, 2023 •

edited

Loading

greyfairer left a comment

greyfairer Apr 5, 2024

greyfairer Apr 5, 2024

greyfairer Apr 5, 2024

greyfairer Apr 5, 2024

greyfairer commented Apr 10, 2024

	this.transformation = new HeaderToField.Key<>();
	this.transformation = new HeaderToField.Value<>();

-    Schema inputSchema = SchemaBuilder.map(SchemaBuilder.STRING_SCHEMA, SchemaBuilder.OPTIONAL_STRING_SCHEMA)
+   Map<String, Object> inputSchema = new HashMap<>();
+   value.put("firstName", "example");
+   value.put("lastName", "user");

	assertEquals(expectedSchema.parameters().size(), 3);
	assertEquals("testing", ((Map<String, String>)actualRecord.value()).get("applicationId"));

fix: implement processMap function to MAP structured data #99

Are you sure you want to change the base?

fix: implement processMap function to MAP structured data #99

Conversation

peacecwz commented May 10, 2023

jcustenborder commented May 10, 2023

peacecwz commented May 10, 2023

peacecwz commented May 10, 2023

peacecwz commented May 10, 2023

peacecwz commented May 25, 2023

harpaj Sep 13, 2023 • edited Loading

Choose a reason for hiding this comment

greyfairer Apr 5, 2024

Choose a reason for hiding this comment

okayhooni commented Oct 20, 2023 • edited Loading

greyfairer left a comment

Choose a reason for hiding this comment

greyfairer Apr 5, 2024

Choose a reason for hiding this comment

greyfairer Apr 5, 2024

Choose a reason for hiding this comment

greyfairer Apr 5, 2024

Choose a reason for hiding this comment

greyfairer Apr 5, 2024

Choose a reason for hiding this comment

greyfairer commented Apr 10, 2024

harpaj Sep 13, 2023 •

edited

Loading

okayhooni commented Oct 20, 2023 •

edited

Loading