KAFKA-4208: Add Record Headers #2772

michaelandrepearce · 2017-03-30T19:27:47Z

As per KIP-82

Adding record headers api to ProducerRecord, ConsumerRecord
Support to convert from protocol to api added Kafka Producer, Kafka Fetcher (Consumer)
Updated MirrorMaker, ConsoleConsumer and scala BaseConsumer
Add RecordHeaders and RecordHeader implementation of the interfaces Headers and Header

Some bits using are reverted to being Java 7 compatible, for the moment until KIP-118 is implemented.

asfbot · 2017-03-30T19:28:52Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/2546/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-03-30T19:28:55Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/2546/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-03-30T19:29:07Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/2550/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-03-30T19:39:07Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/2547/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-03-30T19:39:08Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/2547/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-03-30T19:39:16Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/2551/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-03-30T19:43:48Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/2548/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-03-30T19:43:58Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/2548/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-03-30T19:44:05Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/2552/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-03-30T19:51:00Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/2549/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-03-30T19:51:13Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/2553/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-03-30T19:51:47Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/2549/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-03-30T20:03:18Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/2551/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-03-30T20:03:58Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/2555/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-03-30T20:04:22Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/2551/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-03-30T20:58:30Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/2553/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-03-30T21:01:36Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/2553/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-03-30T21:08:39Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/2557/
Test PASSed (JDK 8 and Scala 2.11).

michaelandrepearce

Review notes, to address small formatting changes, missed in cherry pick merge.

michaelandrepearce · 2017-03-31T07:29:53Z

clients/src/main/java/org/apache/kafka/common/serialization/Serializer.java

@@ -35,7 +35,7 @@
     * @param isKey whether is for key or value
     */
    public void configure(Map<String, ?> configs, boolean isKey);
-
+    


accidental formatting, no need, need to revert.

michaelandrepearce · 2017-03-31T07:30:14Z

clients/src/test/java/org/apache/kafka/clients/consumer/ConsumerRecordTest.java

@@ -42,6 +43,8 @@ public void testOldConstructor() {
        assertEquals(ConsumerRecord.NULL_CHECKSUM, record.checksum());
        assertEquals(ConsumerRecord.NULL_SIZE, record.serializedKeySize());
        assertEquals(ConsumerRecord.NULL_SIZE, record.serializedValueSize());
+        assertEquals(new RecordHeaders(), record.headers());
+


remove extra space

michaelandrepearce · 2017-03-31T07:30:33Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java

+        assertEquals("headerValue2", new String(record.headers().lastHeader("headerKey").value(), Charset.forName("UTF-8")));
+    }
+
+


remove extra un-needed whitespace

michaelandrepearce · 2017-03-31T07:30:54Z

clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerInterceptorsTest.java

@@ -180,7 +180,7 @@ public void testOnAcknowledgementWithErrorChain() {
        assertEquals(2, onErrorAckWithTopicPartitionSetCount);

        // if producer record does not contain partition, interceptor should get partition == -1
-        ProducerRecord<Integer, String> record2 = new ProducerRecord<>("test2", null, 1, "value");
+        ProducerRecord<Integer, String> record2 = new ProducerRecord<>("test2", (Integer) null, 1, "value");


is this needed?

michaelandrepearce · 2017-03-31T07:31:53Z

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

@@ -477,18 +483,23 @@ private static int parseAcks(String acksString) {
                        " to class " + producerConfig.getClass(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG).getName() +
                        " specified in value.serializer");
            }
-
+            


remove accidental, whitespace formatting change.

asfbot · 2017-03-31T07:50:00Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/2567/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-03-31T07:50:02Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/2567/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-03-31T07:50:12Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/2571/
Test FAILed (JDK 8 and Scala 2.11).

michaelandrepearce · 2017-03-31T07:56:22Z

need to rebase, dd71e4a changed test class's breaking FetcherTest on CI build

asfbot · 2017-03-31T09:21:47Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/2569/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-04-28T07:35:15Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3271/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-04-28T07:35:21Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3262/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-04-28T07:35:53Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3266/
Test FAILed (JDK 7 and Scala 2.10).

As per KIP-82 Adding record headers api to ProducerRecord, ConsumerRecord Support to convert from protocol to api added Kafka Producer, Kafka Fetcher (Consumer) Updated MirrorMaker, ConsoleConsumer and scala BaseConsumer Add RecordHeaders and RecordHeader implementation of the interfaces Headers and Header Converting filter iterable into a private static class as per review comments. Updating Java Doc to be more detailed as per review feedback Check key variables passed to methods, to ensure it is not null, and throw if it is, as per Header behaviour when key is null. Refactor duplicated code into private methods. Update ProducerRecord so Iterable<Header> is after V value so is consistent with other areas, in response to review comment. Support headers being able to be availble for serialization/deserialization for use case: https://cwiki.apache.org/confluence/display/KAFKA/A+Case+for+Kafka+Headers Some parts of solution are in-lieu of Java 8 default methods.

@becketqin

Addressing review comments by @becketqin

revert back in Java 7 parts.

@hachikuji

Addressing review comments from @hachikuji

Fix constructor to take Iterable<Header>

Fix conflict post rebase.

asfbot · 2017-04-28T07:45:30Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3263/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-04-28T07:46:00Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3267/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-04-28T07:46:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3272/
Test FAILed (JDK 8 and Scala 2.11).

Enhance Tests, change close() to setReadOnly(), minor other review changes.

michaelandrepearce · 2017-04-28T07:55:25Z

rebased .... again ....

https://s3.amazonaws.com/media-p.slid.es/uploads/403474/images/1927371/conflictRseolutionCartoon.jpg

asfbot · 2017-04-28T08:43:46Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3268/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-04-28T08:43:57Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3264/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-04-28T08:47:23Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3273/
Test PASSed (JDK 8 and Scala 2.11).

hachikuji · 2017-04-29T00:14:16Z

Merge to trunk is imminent. The record header saga is finally drawing to a close. What battles await tomorrow?!

(In all seriousness, thanks for the patch and the persistence with this KIP)

simplesteph · 2017-04-29T03:48:17Z

Sorry I'm late to the party and I'm going to be a pain, but if the headers are making it to the ProducerRecord, they should also make it to the SourceRecord, so that the Connect Framework directly benefits from it.
Also, IMO this would have been a perfection opportunity of adding such headers as a method or something... Right now, many constructors are missing the headers parameter, and the only way to get to add headers is the very dangerous https://github.com/apache/kafka/pull/2772/files#diff-6deeb698c2574836af58336908d76ae5R66 (because of the partition parameter). Just saw this got merged an hour ago...

CF https://cwiki.apache.org/confluence/display/KAFKA/KIP+141+-+ProducerRecord+Interface+Improvements

cc @ijuma @mjsax

michaelandrepearce · 2017-04-29T04:49:12Z

Hi Steph,

You actually add a header in normal usage by calling record.headers.add(key,value), this constructor was added in the record really only aimed at mirror making solutions.

Also adding headers is mostly aimed to be added by interceptors (where record is given) thus mutable at that stage. This is why on produce once sent the headers are made read only.

Yes connect framework isn't done as wasn't covered in this KIP, once this KIP was done I was actially going to raise another KIP immediately to add them (a bit like timestamp was first added to core and then a second kip was raised to add them to connect), but first step it was important to get headers added at the core level. We actually have the code for that ready so hold on :)

simplesteph · 2017-04-29T04:55:22Z

That helps thanks !

michaelandrepearce · 2017-04-29T07:23:05Z

@simplesteph
Was going to raise all this in a week or so time, but as we had the code ready and obviously you have a need thus raised the point re connect, i have raised it all now.

Here's KIP
https://cwiki.apache.org/confluence/display/KAFKA/KIP-145+-+Expose+Record+Headers+in+Kafka+Connect

Here's PR (which we had the code for)
#2942

Obviously first needs KIP discussion and vote. So please feel free to get active in that space supporting that :)

mjsax · 2017-04-29T17:11:57Z

@simplesteph See https://cwiki.apache.org/confluence/display/KAFKA/KIP-82+-+Add+Record+Headers for details about this work. You can just update your own KIP accordingly.

Update upgrade.html Raising this now, as KIP-118 is pulled from release as such submitting this without java 8 changes. As per remaining review comment from #2772, updating the upgrade notes. Author: Michael André Pearce <michael.andre.pearce@me.com> Author: Michael Andre Pearce <Michael.Andre.Pearce@me.com> Reviewers: Jiangjie Qin <becket.qin@gmail.com>, Ismael Juma <ismael@juma.me.uk> Closes #2991 from michaelandrepearce/KIP-82

michaelandrepearce force-pushed the KIP-82 branch from f70c094 to 6245945 Compare March 30, 2017 19:37

michaelandrepearce force-pushed the KIP-82 branch from 6245945 to 02e982a Compare March 30, 2017 19:42

michaelandrepearce force-pushed the KIP-82 branch from 02e982a to 04f81fa Compare March 30, 2017 19:48

michaelandrepearce force-pushed the KIP-82 branch from 04f81fa to 49cfa31 Compare March 30, 2017 20:00

michaelandrepearce force-pushed the KIP-82 branch from 49cfa31 to a843f7c Compare March 30, 2017 20:10

michaelandrepearce commented Mar 31, 2017

View reviewed changes

michaelandrepearce force-pushed the KIP-82 branch from cdf4cf1 to b4a0365 Compare March 31, 2017 08:26

michaelandrepearce added 6 commits April 28, 2017 08:38

KAFKA-4208: Add Record Headers

f35a932

Addressing review comments by @becketqin

KAFKA-4208: Add Record Headers

244c211

revert back in Java 7 parts.

KAFKA-4208: Add Record Headers

a4ecf1c

Addressing review comments from @hachikuji

KAFKA-4208: Add Record Headers

aebe286

Fix constructor to take Iterable<Header>

KAFKA-4208: Add Record Headers

cb6d68a

Fix conflict post rebase.

michaelandrepearce force-pushed the KIP-82 branch from cfd7796 to b75f939 Compare April 28, 2017 07:42

KAFKA-4208: Add Record Headers

0d3a241

Enhance Tests, change close() to setReadOnly(), minor other review changes.

michaelandrepearce force-pushed the KIP-82 branch from b75f939 to 0d3a241 Compare April 28, 2017 07:48

asfgit closed this in 6185bc0 Apr 29, 2017

michaelandrepearce deleted the KIP-82 branch May 7, 2017 16:23

michaelandrepearce mentioned this pull request May 7, 2017

KAFKA-4208: Add Record Headers #2991

Closed

		assertEquals("headerValue2", new String(record.headers().lastHeader("headerKey").value(), Charset.forName("UTF-8")));
		}

KAFKA-4208: Add Record Headers #2772

KAFKA-4208: Add Record Headers #2772

Conversation

michaelandrepearce commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

asfbot commented Mar 30, 2017

michaelandrepearce left a comment

Choose a reason for hiding this comment

michaelandrepearce Mar 31, 2017

Choose a reason for hiding this comment

michaelandrepearce Mar 31, 2017

Choose a reason for hiding this comment

michaelandrepearce Mar 31, 2017

Choose a reason for hiding this comment

michaelandrepearce Mar 31, 2017

Choose a reason for hiding this comment

michaelandrepearce Mar 31, 2017

Choose a reason for hiding this comment

asfbot commented Mar 31, 2017

asfbot commented Mar 31, 2017

asfbot commented Mar 31, 2017

michaelandrepearce commented Mar 31, 2017

asfbot commented Mar 31, 2017

asfbot commented Apr 28, 2017

asfbot commented Apr 28, 2017

asfbot commented Apr 28, 2017

asfbot commented Apr 28, 2017

asfbot commented Apr 28, 2017

asfbot commented Apr 28, 2017

michaelandrepearce commented Apr 28, 2017

asfbot commented Apr 28, 2017

asfbot commented Apr 28, 2017

asfbot commented Apr 28, 2017

hachikuji commented Apr 29, 2017

simplesteph commented Apr 29, 2017 • edited

michaelandrepearce commented Apr 29, 2017

simplesteph commented Apr 29, 2017

michaelandrepearce commented Apr 29, 2017

mjsax commented Apr 29, 2017

simplesteph commented Apr 29, 2017 •

edited