KIP-145: Add SMTs, HeaderFrom, DropHeaders and InsertHeader by tombentley · Pull Request #9549 · apache/kafka

tombentley · 2020-11-03T16:09:55Z

These SMTs were originally specified in KIP-145 but never implemented at the time.

HeaderTo is not included since its original specification doesn't deal with the fact that there can be >1 header with the same name, but a field can only have a single value (while you could use an array, that doesn't work if the headers for the given name had different schemas).

tombentley · 2020-11-03T16:13:04Z

@kkonstantine would you be able to review this? These SMTs were originally specified in KIP145 but never implemented. It seems they were forgotten about.

@C0urante maybe you'd also like to review?

C0urante · 2020-11-09T19:04:05Z

Thanks for reaching out @tombentley! Happy to take a look.

For reference, maybe you could add a link to https://cwiki.apache.org/confluence/display/KAFKA/KIP-145+-+Expose+Record+Headers+in+Kafka+Connect#KIP145ExposeRecordHeadersinKafkaConnect-Transformations to the description?

Will begin reviewing shortly.

C0urante

Looking pretty good! Will check out the tests after the core functional changes are settled.

C0urante · 2020-11-09T19:10:07Z

Might make sense to refer to the MOVE.name and COPY.name fields declared in the Operation enum below instead of using the literal "move" and "copy" strings in this section.

MOVE.name is not a compile time constant (at least the compiler doesn't see it that way). I factored out constants to avoid the duplication of literals.

C0urante · 2020-11-09T19:12:07Z

Why duplicate headers here? According to the Header class's Javadocs, the collection should be mutable.

I wasn't completely sure whether a transformation was allowed to mutate headers, since it has to create a new ConnectRecord. Take this code from WorkerSourceTask as an example:

final SourceRecord record = transformationChain.apply(preTransformRecord); final ProducerRecord<byte[], byte[]> producerRecord = convertTransformedRecord(record); if (producerRecord == null || retryWithToleranceOperator.failed()) { counter.skipRecord(); commitTaskRecord(preTransformRecord, null); continue; }

See how preTransformRecord can be used after the transformation chain has been applied? I think it would be incorrect for commitTaskRecord to commit the original record but with headers which had been mutated by transformations, right?

Ah, that's fair. Does seem to be the pattern followed by the other out-of-the-box transformations as well; probably best to continue to follow that pattern.

I'm a little unnerved by this though, since as far as I can tell it's not publicly documented and so it's possible people writing their own transformations may be violating this implicit rule.

Out of scope, so I've filed KAFKA-10720 to track the need for possible documentation improvements.

C0urante · 2020-11-09T19:16:34Z

This looks fairly expensive to perform for every record. Do you think it might make sense to perform some caching, similarly to what's done in the ReplaceField transform?

C0urante · 2020-11-09T19:43:32Z

Same question here RE: duplication

C0urante · 2020-11-09T19:46:38Z

Same question here RE: duplication

tombentley · 2020-11-11T09:09:52Z

@C0urante thanks for the review, some really helpful comments there.

C0urante · 2020-11-13T18:56:30Z

Little silly to run this for every iteration of the parameterized test 😛
I'm guessing there isn't an easy way to run this only once?

Yeah, parameterized tests do have their faults. AFAIK the only way to handle this using Junit 4 is to have two tests. You can arrange to have both tests in the same source file so both parameterised and non-parameterised tests are still easily discoverable (e.g. using Enclosed), but I've never seen that done in practice and it looks rather fiddly. Since these tests execute pretty quickly, I'm inclined to think we should just live with it. The real solution might just be to adopt Junit 5.

C0urante

LGTM, a few small comments but none of them are blockers. Thanks Tom!

tombentley · 2020-11-16T09:25:19Z

@kkonstantine or perhaps @rhauch please could one of you take a look?

tombentley · 2020-11-26T17:40:19Z

@kkonstantine, @rhauch did you get a chance to look at this?

mimaison

Thanks or the PR! Good to see these SMTs finally being added.
I've made a first quick pass (haven't looked at HeaderFrom yet) and left a few suggestions

mimaison · 2021-03-05T12:31:05Z

Unused import

mimaison · 2021-03-05T12:34:55Z

We don't have this transformation!

mimaison · 2021-03-05T14:09:33Z

Because Headers is a LinkedList, remove() has to iterate the whole list each time. I wonder if we could instead start from an empty headers list and add the headers not being removed?

mimaison · 2021-03-05T14:09:42Z

nit, extra line

mimaison · 2021-03-05T14:09:48Z

nit, extra line

mimaison · 2021-03-05T14:29:55Z

Should we enforce these fields are not null?

I guess that makes sense @mimaison, but then for consistency shouldn't we make DropHeaders.headers, HeaderFrom.headers and HeaderFrom.fields reject empty lists?

mimaison · 2021-03-05T14:32:51Z

Should we reuse MOVE_OPERATION and COPY_OPERATION here?

tombentley · 2021-04-07T11:23:21Z

@mimaison I went ahead and enforce non-nullness and non-emptiness, if you want to do another pass?

mimaison · 2021-04-15T09:32:50Z

Even though I don't think it's reachable by users, should we have a message here?

It is impossible due to the ConfigDef.ValidString.in(MOVE_OPERATION, COPY_OPERATION), so this is really an assertion failure. The line number in the stacktrace would be enough to track it down if it ever did happen due to a later refactoring, so imho an error message is of no value. But I'm happy to add one if you like.

Ok, that's fair enough. Thanks

mimaison

Thanks Tom, it looks good!

mimaison · 2021-04-15T10:00:15Z

Not sure why the build had failed, I've rekicked it.

…Header These SMTs were originally specified in KIP-145 but never implemented at the time. HeaderTo is not included since its original specification doesn't deal with the fact that there can be >1 header with the same name, but a field can only have a single value (which could be an array, but not if the headers for the given name had different schemas).

tombentley · 2021-04-16T14:09:21Z

Failing tests are those reported in KAFKA-12629, KAFKA-12284 and KAFKA-9295:

Build / JDK 11 and Scala 2.13 / testCreateClusterAndCreateAndManyTopicsWithManyPartitions() – kafka.server.RaftClusterTest32s
Build / JDK 8 and Scala 2.12 / testReplication() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationSSLTest1m 32s
Build / JDK 8 and Scala 2.12 / testReplicationWithEmptyPartition() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationSSLTest1m 26s
Build / JDK 8 and Scala 2.12 / testOneWayReplicationWithAutoOffsetSync() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationSSLTest1m 23s
Build / JDK 8 and Scala 2.12 / testReplication() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTest1m 27s
Build / JDK 8 and Scala 2.12 / testReplicationWithEmptyPartition() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTest1m 16s
Build / JDK 8 and Scala 2.12 / shouldInnerJoinMultiPartitionQueryable – org.apache.kafka.streams.integration.KTableKTableForeignKeyInnerJoinMultiIntegrationTest

All these pass using the relevant JDKs on my machine. Merging to trunk.

C0urante reviewed Nov 9, 2020

View reviewed changes

C0urante reviewed Nov 13, 2020

View reviewed changes

C0urante approved these changes Nov 13, 2020

View reviewed changes

tombentley force-pushed the KIP-145-add-missing-smt branch from e426dbf to 1a79ffd Compare November 26, 2020 17:40

mimaison requested changes Mar 5, 2021

View reviewed changes

tombentley force-pushed the KIP-145-add-missing-smt branch from 1a79ffd to a211f1e Compare April 7, 2021 11:22

tombentley mentioned this pull request Apr 12, 2021

KAFKA-8863 - Add InsertHeader and DropHeader transforms for connect #7284

Closed

3 tasks

mimaison reviewed Apr 15, 2021

View reviewed changes

mimaison approved these changes Apr 15, 2021

View reviewed changes

tombentley added 5 commits April 16, 2021 09:14

Review comments

3233c59

Docs

9e8ef0b

Review comments

6e75b8c

Convert tests to junit5

c7ef2dc

tombentley force-pushed the KIP-145-add-missing-smt branch from a211f1e to c7ef2dc Compare April 16, 2021 08:15

tombentley merged commit f8f1769 into apache:trunk Apr 16, 2021

tombentley deleted the KIP-145-add-missing-smt branch April 16, 2021 14:11

Conversation

tombentley commented Nov 3, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

tombentley commented Nov 3, 2020

Uh oh!

C0urante commented Nov 9, 2020

Uh oh!

C0urante 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!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

tombentley commented Nov 11, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

C0urante left a comment

Choose a reason for hiding this comment

Uh oh!

tombentley commented Nov 16, 2020

Uh oh!

tombentley commented Nov 26, 2020

Uh oh!

mimaison 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!

tombentley commented Apr 7, 2021

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!

mimaison left a comment

Choose a reason for hiding this comment

Uh oh!

mimaison commented Apr 15, 2021

Uh oh!

tombentley commented Apr 16, 2021

tombentley commented Nov 3, 2020 •

edited

Loading