KAFKA-8863 - Add InsertHeader and DropHeader transforms for connect

[alozano3]

The SMTs InsertHeader and DropHeaders described in KIP-145 have been implemented, extending the work done in PR4319.

Unit tests are added for InsertHeader and ConnectHeaders, covering the different types of Header (int8, int16, int32, int64, float32, float64, Time, Date and Timestamp).

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[alozano3]

Inside the method consumeNextToken() of the Values class, the string being parsed is tokenized by the character ":". So when I insert a header of the type Time with the pattern "HH:mm:ss.SSS'Z'" the string gets tokenized instead of remaining as a whole.

The consequence is that the string is treated as different integer values and is never parsed as a whole date inside the method parse(Parser parser, boolean embedded).

As a workaround I insert the value escaped (e.g: "14\\:34\\:22.222Z"), and then I replace the characters "\" by "", because in order to be aprsed as adate, the length of the string is checked.

Could you give some advice on this please??

[alozano3]

retest this please

[alozano3]

retest this please

[alozano3]

There is a flaky test org.apache.kafka.trogdor.coordinator.CoordinatorTest.testTaskRequestWithOldStartMsGetsUpdated

org.junit.runners.model.TestTimedOutException: test timed out after 120000 milliseconds at java.base@11.0.1/jdk.internal.misc.Unsafe.park(Native Method) at java.base@11.0.1/java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:234) at java.base@11.0.1/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2123) at java.base@11.0.1/java.util.concurrent.ThreadPoolExecutor.awaitTermination(ThreadPoolExecutor.java:1454) at java.base@11.0.1/java.util.concurrent.Executors$DelegatedExecutorService.awaitTermination(Executors.java:709) at app//org.apache.kafka.trogdor.rest.JsonRestServer.waitForShutdown(JsonRestServer.java:158) at app//org.apache.kafka.trogdor.agent.Agent.waitForShutdown(Agent.java:123) at app//org.apache.kafka.trogdor.common.MiniTrogdorCluster.close(MiniTrogdorCluster.java:285) at app//org.apache.kafka.trogdor.coordinator.CoordinatorTest.testTaskRequestWithOldStartMsGetsUpdated(CoordinatorTest.java:596) at java.base@11.0.1/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at java.base@11.0.1/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at java.base@11.0.1/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.base@11.0.1/java.lang.reflect.Method.invoke(Method.java:566) at app//org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59) at app//org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12) at app//org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56) at app//org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17) at app//org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:288) at app//org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:282) at java.base@11.0.1/java.util.concurrent.FutureTask.run(FutureTask.java:264) at java.base@11.0.1/java.lang.Thread.run(Thread.java:834)

[alozano3]

Another flaky test:
org.apache.kafka.connect.integration.ExampleConnectIntegrationTest.testSourceConnector

org.apache.kafka.connect.errors.DataException: Insufficient records committed by connector simple-conn in 30000 millis. Records expected=2000, actual=79 at org.apache.kafka.connect.integration.ConnectorHandle.awaitCommits(ConnectorHandle.java:191) at org.apache.kafka.connect.integration.ExampleConnectIntegrationTest.testSourceConnector(ExampleConnectIntegrationTest.java:183) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

[alozano3]

retest this please

[alozano3]

@rhauch for review.

[alozano3]

Should I just insert only string schema to the header instead of trying to parse the value in order to set the proper schema??

[rhauch]

Thanks for the contribution, @alozano3! Very nice work. There are a few behavioral changes that we can make, but overall this is very close.

[alozano3]

@rhauch What do you think about the issue with the Time and Timestamp values that I previously posted?
The string gets tokenized in the method _Values.consumeNextToken because it contains the character ':' so it is treated like a key:value field. So I have to put the header value escaping the colons "14\\:34\\:22.222Z" and then delete the escape characters in the Values.parse method:

// Check for a date, time, or timestamp ...
token = token.replace("\\", "");
int tokenLength = token.length();

[rhauch]

Thanks, @alozano3. Looks great, though I have two pretty easy suggestions below.

[rhauch]

Thanks for the quick turnaround, @alozano3. A few more minor suggestions.

[rhauch]

Retest this, please.

[rhauch]

@alozano3 any update to my questions/comments?

[alozano3]

Sure! I have implemented and answered some of your advices but I didn't have much time yesterday to have it all. Hopefully next week, I will go for it.

[rhauch]

@alozano3 we're getting close to code freeze, so the sooner the better.

[rhauch]

Thanks for the PR, @alozano3. The SMT implementations look pretty good, but unfortunately any changes to ConfigDef will require a KIP. My recommendation is to avoid making any changes to ConfigDef and keep the validator implementation specific to these SMTs.

[alozano3]

Any update?

[alozano3]

Thanks for the PR, @alozano3. The SMT implementations look pretty good, but unfortunately any changes to ConfigDef will require a KIP. My recommendation is to avoid making any changes to ConfigDef and keep the validator implementation specific to these SMTs.

The change is done. Please review the changes.
Thanks.

[tombentley]

@alozano3 I have PR #9549 open for these SMTs (plus HeaderFrom) and was hoping to merge them this week pending a 2nd pass from @mimaison.

[alozano3]

@alozano3 I have PR #9549 open for these SMTs (plus HeaderFrom) and was hoping to merge them this week pending a 2nd pass from @mimaison.

Great because I have not received feedback for over a year..

[alozano3]

I close this PR because #9549 has been merged.