KAFKA-4935: Deprecate client checksum API and compute lazy partial checksum for magic v2 #3123

hachikuji · 2017-05-23T07:22:11Z

No description provided.

…ecksum for magic v2

hachikuji · 2017-05-23T07:22:29Z

hachikuji · 2017-05-23T16:14:36Z

retest this please

asfbot · 2017-05-23T17:39:25Z

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

asfbot · 2017-05-23T17:39:35Z

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

ijuma · 2017-05-23T20:19:29Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecord.java

@@ -112,7 +114,7 @@ public ConsumerRecord(String topic,
                          long offset,
                          long timestamp,
                          TimestampType timestampType,
-                          long checksum,
+                          Long checksum,


Adding a note here in case others are wondering: this is OK because this constructor was added during this release cycle.

ijuma

Thanks for the PR. Looks good overall, just a few minor comments.

ijuma · 2017-05-23T20:29:06Z

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecord.java

+     * @deprecated As of Kafka 0.11.0. Because of the potential for message format conversion on the broker, the
+     *             checksum returned by the broker may not match what was computed by the producer.
+     *             It is therefore unsafe to depend on this checksum for end-to-end delivery guarantees. Additionally,
+     *             message format v2 does not support a record-level checksum. To maintain compatibility, a partial


Maybe mention that record-level checksums were removed for performance reasons.

ijuma · 2017-05-23T20:32:56Z

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

+    private volatile Long checksum;
+
+    public RecordMetadata(TopicPartition topicPartition, long baseOffset, long relativeOffset, long timestamp,
+                          Long checksum, int serializedKeySize, int serializedValueSize) {


long -> Long is a binary incompatible change.

ijuma · 2017-05-23T20:33:21Z

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

@@ -54,15 +57,7 @@ private RecordMetadata(TopicPartition topicPartition, long offset, long timestam

    @Deprecated
    public RecordMetadata(TopicPartition topicPartition, long baseOffset, long relativeOffset) {


This is deprecated. Shall we remove it or undeprecate it?

Seems we deprecated in 0.10.0. Maybe it is safe to remove now?

Yeah, I was thinking it was fine to remove it since users should not create an instance of this apart from tests and it's been deprecated for a few release cycles. Or if we think the method is useful, then it would make sense undeprecate it.

Makes sense. Let's remove it.

ijuma · 2017-05-23T20:34:47Z

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

    public long checksum() {
+        if (checksum == null)
+            this.checksum = DefaultRecordBatch.computePartialChecksum(timestamp, serializedKeySize, serializedValueSize);


We should probably add a comment here and ConsumerRecord explaining that this will be null for message format >= 2 and hence why we call DefaultRecordBatch.

ijuma · 2017-05-23T20:35:49Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/FutureRecordMetadata.java

@@ -96,14 +96,10 @@ RecordMetadata valueOrError() throws ExecutionException {
            return value();
    }

-    long checksum() {
+    Long checksum() {


Should this be checksumOrNull?

ijuma · 2017-05-23T20:37:05Z

clients/src/main/java/org/apache/kafka/common/record/DefaultRecord.java

    }

    /**
     * Write the record to `out` and return its crc.
     */
-    public static long writeTo(ByteBuffer out,
+    public static int writeTo(ByteBuffer out,


Comment above needs to be updated.

ijuma · 2017-05-23T20:37:42Z

clients/src/main/java/org/apache/kafka/common/record/DefaultRecordBatch.java

@@ -493,4 +496,13 @@ public void remove() {
        }

    }
+
+    public static long computePartialChecksum(long timestamp, int serializedKeySize, int serializedValueSize) {


Is there a reason why this is not in DefaultRecord?

ijuma · 2017-05-23T20:41:17Z

clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java

@@ -344,7 +344,7 @@ private int writeLegacyCompressedWrapperHeader() {
        return writtenCompressed;
    }

-    private long appendWithOffset(long offset, boolean isControlRecord, long timestamp, ByteBuffer key,
+    private Long appendWithOffset(long offset, boolean isControlRecord, long timestamp, ByteBuffer key,


Maybe add a comment stating the return value.

ijuma · 2017-05-23T20:42:45Z

clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java

-        recordWritten(offset, timestamp, DefaultRecord.sizeInBytes(offsetDelta, timestampDelta, key, value, headers));
-        return crc;
+        int sizeInBytes = DefaultRecord.writeTo(appendStream, offsetDelta, timestampDelta, key, value, headers);
+        recordWritten(offset, timestamp, sizeInBytes);


Nice improvement.

apurvam

The basic changes look good to me. Not much you can do differently IMO. It would be good to have some tests to make sure that the checksum methods on RecordMetadata (on the producer side) and the ConsumerRecord (on the consume side) return what's expected.

hachikuji · 2017-05-24T05:06:06Z

@ijuma @apurvam Comments addressed. Please take another look.

asfbot · 2017-05-24T06:30:11Z

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

asfbot · 2017-05-24T06:37:35Z

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

asfbot · 2017-05-24T17:18:26Z

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

asfbot · 2017-05-24T17:18:37Z

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

ijuma

LGTM, thanks. Before merging I did a few trivial changes:

Mentioned removed constructor in upgrade notes (we already had a line for that, so just mentioned the class name).
Avoided some deprecation warnings by using Long.valueOf so that the right constructor is chosen.
Added @SuppressWarnings("deprecation") in a few places.

ijuma · 2017-05-25T06:40:13Z

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

-             timestamp, checksum, serializedKeySize, serializedValueSize);
+    public RecordMetadata(TopicPartition topicPartition, long baseOffset, long relativeOffset, long timestamp,
+                          long checksum, int serializedKeySize, int serializedValueSize) {
+        this(topicPartition, baseOffset, relativeOffset, timestamp, (Long) checksum, serializedKeySize,


I changed the cast to Long.valueOf.

ijuma · 2017-05-25T07:03:15Z

clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java

        Struct keyStruct = type.recordKey();
        ByteBuffer key = ByteBuffer.allocate(keyStruct.sizeOf());
        keyStruct.writeTo(key);
        key.flip();
        return appendWithOffset(nextSequentialOffset(), true, timestamp, key, value, Record.EMPTY_HEADERS);
    }

-    public long appendEndTxnMarker(long timestamp, EndTransactionMarker marker) {
+    public Long appendEndTxnMarker(long timestamp, EndTransactionMarker marker) {


I documented the return type above.

ijuma · 2017-05-25T07:19:27Z

clients/src/main/java/org/apache/kafka/common/record/DefaultRecord.java

     */
-    public static long writeTo(DataOutputStream out,
+    public static int writeTo(DataOutputStream out,


I fixed the alignment here and the other method in this class.

…ecksum for magic v2 Author: Jason Gustafson <jason@confluent.io> Reviewers: Apurva Mehta <apurva@confluent.io>, Ismael Juma <ismael@juma.me.uk> Closes #3123 from hachikuji/KAFKA-4935

KAFKA-4935: Deprecate client checksum API and compute lazy partial ch…

8b161c4

…ecksum for magic v2

ijuma reviewed May 23, 2017

View reviewed changes

apurvam reviewed May 23, 2017

View reviewed changes

hachikuji force-pushed the KAFKA-4935 branch from 94d182b to 02c29e0 Compare May 24, 2017 05:04

Address review comments and add a few test cases

07d0788

hachikuji force-pushed the KAFKA-4935 branch from 02c29e0 to 07d0788 Compare May 24, 2017 05:04

hachikuji changed the title ~~KAFKA-4935 [WIP]: Deprecate client checksum API and compute lazy partial checksum for magic v2~~ KAFKA-4935: Deprecate client checksum API and compute lazy partial checksum for magic v2 May 24, 2017

Add newline to RecordMetadataTest

23a38ff

Remove deprecated (as of 0.10.0) RecordMetadata constructor

2394f5c

ijuma approved these changes May 25, 2017

View reviewed changes

asfgit closed this in cea319a May 25, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-4935: Deprecate client checksum API and compute lazy partial checksum for magic v2 #3123

KAFKA-4935: Deprecate client checksum API and compute lazy partial checksum for magic v2 #3123

hachikuji commented May 23, 2017

hachikuji commented May 23, 2017

hachikuji commented May 23, 2017

asfbot commented May 23, 2017

asfbot commented May 23, 2017

ijuma May 23, 2017

ijuma left a comment

ijuma May 23, 2017

ijuma May 23, 2017

ijuma May 23, 2017

hachikuji May 24, 2017

ijuma May 24, 2017

hachikuji May 24, 2017

ijuma May 23, 2017

ijuma May 23, 2017

ijuma May 23, 2017

ijuma May 23, 2017

ijuma May 23, 2017

ijuma May 23, 2017

apurvam left a comment

hachikuji commented May 24, 2017

asfbot commented May 24, 2017

asfbot commented May 24, 2017

asfbot commented May 24, 2017

asfbot commented May 24, 2017

ijuma left a comment

ijuma May 25, 2017

ijuma May 25, 2017

ijuma May 25, 2017

		@@ -54,15 +57,7 @@ private RecordMetadata(TopicPartition topicPartition, long offset, long timestam

		@Deprecated
		public RecordMetadata(TopicPartition topicPartition, long baseOffset, long relativeOffset) {

KAFKA-4935: Deprecate client checksum API and compute lazy partial checksum for magic v2 #3123

KAFKA-4935: Deprecate client checksum API and compute lazy partial checksum for magic v2 #3123

Conversation

hachikuji commented May 23, 2017

hachikuji commented May 23, 2017

hachikuji commented May 23, 2017

asfbot commented May 23, 2017

asfbot commented May 23, 2017

Choose a reason for hiding this comment

ijuma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apurvam left a comment

Choose a reason for hiding this comment

hachikuji commented May 24, 2017

asfbot commented May 24, 2017

asfbot commented May 24, 2017

asfbot commented May 24, 2017

asfbot commented May 24, 2017

ijuma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment