[KAFKA-8522] Implementing proposal as outlined in KIP-534 #7600

ConcurrencyPractitioner · 2019-10-27T17:22:57Z

Trying to resolve long running issues we have with transactional marker and tombstone retention.

ConcurrencyPractitioner · 2019-11-17T03:03:51Z

@junrao @hachikuji Have finished the initial ground work. Will be glad to get some comments.
Might be a few bugs left which I need to tackle.

ConcurrencyPractitioner · 2019-11-18T22:12:02Z

Retest this please.

ConcurrencyPractitioner · 2019-11-22T00:43:18Z

@junrao @hachikuji Fixed the issue. Ready for review.

ConcurrencyPractitioner · 2019-11-30T21:18:13Z

@hachikuji @guozhangwang Have any comments?

junrao

@ConcurrencyPractitioner : Thanks for the PR. Looks good overall. Left a few comments below.

junrao · 2019-11-18T02:55:31Z

core/src/main/scala/kafka/log/LogCleaner.scala

@@ -643,7 +643,8 @@ private[log] class Cleaner(val id: Int,
                             maxLogMessageSize: Int,
                             transactionMetadata: CleanedTransactionMetadata,
                             lastRecordsOfActiveProducers: Map[Long, LastRecord],
-                             stats: CleanerStats): Unit = {
+                             stats: CleanerStats,
+                             tombstoneRetentionMs: Long): Unit = {


Could we add the javadoc for tombstoneRetentionMs?

junrao · 2019-12-02T01:12:31Z

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

@@ -431,6 +431,16 @@ public LegacyRecord outerRecord() {
            return record;
        }

+        @Override
+        public long deleteHorizonMs() {
+            return -1L;


Could we use RecordBatch.NO_TIMESTAMP here too?

junrao · 2019-12-02T01:19:32Z

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

+    public long deleteHorizonMs() {
+        if (isDeleteHorizonSet()) {
+            return firstTimestamp();
+        }


No need for {} for single line statement.

junrao · 2019-12-02T01:20:29Z

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

        return attributes;
    }

+    public static void writeEmptyHeader(ByteBuffer buffer,
+            byte magic,


The existing indentation is correct.

junrao · 2019-12-02T01:23:52Z

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

+        public long deleteHorizonMs() {
+            if (isDeleteHorizonSet()) {
+                return super.loadFullBatch().deleteHorizonMs();
+            }


No need for {} for single line statement.

junrao · 2019-12-02T01:36:36Z

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

+            if (firstTimestamp == null) {
+                if (isDeleteHorizonSet()) {
+                    firstTimestamp = deleteHorizonMs;
+                } else {


No need for {} for single line statement.

junrao · 2019-12-02T02:05:29Z

core/src/main/scala/kafka/log/LogCleaner.scala

-      val isRetainedValue = record.hasValue || retainDeletes
+      val isRetainedValue = record.hasValue ||
+                            (!batch.isDeleteHorizonSet() ||
+                             time.milliseconds() < batch.deleteHorizonMs)


We probably should use the new condition only when the record batch is of the latest message format. Otherwise, we could continue to use the old condition.

Oh, I ran tests, and spotbug main reported dodgy code when I checked if the version was less than the most recent one. Apparently, by this point, we already know that the batch is the latest version (v_2) anyways, so we don't need to check for this.

Hmm, how do we know at this point that batch is of V2? It seems that it could be any version.

Alright, will push this to git then. Can't find exactly where it is checked.

junrao · 2019-12-02T02:15:46Z

core/src/main/scala/kafka/log/LogCleaner.scala

+
+        // check that the control batch has been emptied of records
+        // if not, then we do not set a delete horizon until that is true
+        if (batch.isControlBatch() && transactionMetadata.onControlBatchRead(batch))


Hmm, we already called transactionMetadata.onControlBatchRead(batch) in shouldDiscardBatch(). So, we probably don't want to call it again on the same batch. Instead, we want to pass along the return value to retrieveDeleteHorizon().

Yep, working on this. Will address this soon.

junrao · 2019-12-02T02:21:35Z

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

-                if (writeOriginalBatch) {
+                // we check if the delete horizon should be set to a new value
+                // in which case, we need to reset the base timestamp and overwrite the timestamp deltas
+                if (writeOriginalBatch && !shouldSetDeleteHorizon) {


We only need to write the DeleteHorizonTime if the batch contains a tombstone or a control record. This increases the chance for more optimized writeOriginalBatch case to happen.

junrao · 2019-12-02T02:23:38Z

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

- *  -------------------------------------------------------------------------------------------------
- *  | Unused (6-15) | Control (5) | Transactional (4) | Timestamp Type (3) | Compression Type (0-2) |
- *  -------------------------------------------------------------------------------------------------
+ *  ---------------------------------------------------------------------------------------------------------------------------


Could we add some doc on how Delete Horizon flag is being used?

junrao · 2019-12-02T21:33:37Z

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

                    batch.writeTo(bufferOutputStream);
                    filterResult.updateRetainedBatchMetadata(batch, retainedRecords.size(), false);
                } else {
-                    MemoryRecordsBuilder builder = buildRetainedRecordsInto(batch, retainedRecords, bufferOutputStream);
+                    MemoryRecordsBuilder builder = buildRetainedRecordsInto(batch, retainedRecords, bufferOutputStream, firstTimestamp);


We probably only want to set the DeleteHorizonTime if the batch contains tombstone.

Yeah, I missed that. Should've also checked in the else part of the code as well.

junrao · 2019-12-02T21:34:07Z

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

@@ -159,6 +159,7 @@ private static FilterResult filterTo(TopicPartition partition, Iterable<MutableR
        for (MutableRecordBatch batch : batches) {
            long maxOffset = -1L;
            BatchRetention batchRetention = filter.checkBatchRetention(batch);
+            final long firstTimestamp = filter.retrieveDeleteHorizon(batch);


firstTimestamp => deleteHorizonMs?

junrao · 2019-12-02T21:34:25Z

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

            boolean writeOriginalBatch = true;
+            boolean shouldSetDeleteHorizon = !batch.isDeleteHorizonSet() && firstTimestamp != RecordBatch.NO_TIMESTAMP;
+            boolean containsTombstones = false;


containsTombstones => containsTombstonesOrMarker?

ConcurrencyPractitioner · 2019-12-03T15:39:09Z

@junrao All comments have been addressed.

junrao

@ConcurrencyPractitioner : Thanks for the updated patch. A few more comments below. Also, could we add a test for the new logic?

junrao · 2019-12-04T23:22:22Z

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

+        }
+
+        @Override
+        public boolean isDeleteHorizonSet() {


isDeleteHorizonSet => deleteHorizonSet ?

junrao · 2019-12-04T23:30:19Z

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

@@ -557,6 +577,16 @@ public long baseOffset() {
            return loadFullBatch().baseOffset();
        }

+        @Override
+        public long deleteHorizonMs() {
+            return loadFullBatch().deleteHorizonMs();


Since this is a legacy record, deleteHorizonMs is never going to be set. It seems we can avoid loading the full batch? Ditto in isDeleteHorizonSet below.

junrao · 2019-12-04T23:37:09Z

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

+        @Override
+        public long deleteHorizonMs() {
+            if (isDeleteHorizonSet())
+                return super.loadFullBatch().deleteHorizonMs();


Hmm, why do we need to call loadFullBatch()? Could we just call loadBatchHeader()?

junrao · 2019-12-05T21:44:55Z

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

@@ -159,6 +159,7 @@ private static FilterResult filterTo(TopicPartition partition, Iterable<MutableR
        for (MutableRecordBatch batch : batches) {
            long maxOffset = -1L;
            BatchRetention batchRetention = filter.checkBatchRetention(batch);
+            long deleteHorizonMs = filter.retrieveDeleteHorizon(batch);


Hmm, it seems that we can just get deleteHorizonMs from batch directly, instead of through filter.

Yeah, I get what you are saying. Just want to point out that this method is still necessary when the delete horizon is not set.

junrao · 2019-12-05T22:18:08Z

core/src/main/scala/kafka/log/LogCleaner.scala

@@ -785,7 +801,14 @@ private[log] class Cleaner(val id: Int,
       *   2) The message doesn't has value but it can't be deleted now.
       */
      val latestOffsetForKey = record.offset() >= foundOffset
-      val isRetainedValue = record.hasValue || retainDeletes
+      val isLatestVersion = batch.magic() < RecordBatch.MAGIC_VALUE_V2
+      val shouldRetainDeletes = isLatestVersion match {


For boolean value, we can just use if/else instead of match/case.

junrao · 2019-12-05T22:41:31Z

core/src/main/scala/kafka/log/LogCleaner.scala

    } else {
      val canDiscardBatch = transactionMetadata.onBatchRead(batch)
-      canDiscardBatch
+      (canDiscardBatch, false)


If a control batch is empty, we need to check whether the batch should be retained based on deleteHorizonMs similar to what's in shouldRetainRecord. We can ignore retainTxnMarkers since a control batch is always in V2.

Also, it's a bit awkward to have to return 2 values. Maybe it's simpler to just return canDiscardControlBatch and let the caller decide how the batch should be retained.

junrao · 2019-12-05T22:49:36Z

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

            boolean writeOriginalBatch = true;
+            boolean shouldSetDeleteHorizon = !batch.isDeleteHorizonSet() && deleteHorizonMs != RecordBatch.NO_TIMESTAMP;


Hmm, a more intuitive check is probably !batch.isDeleteHorizonSet() && messageFormat >= V2.

junrao · 2019-12-05T22:58:45Z

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

@@ -250,6 +304,7 @@ private static MemoryRecordsBuilder buildRetainedRecordsInto(RecordBatch origina
                originalBatch.compressionType(), timestampType, baseOffset, logAppendTime, originalBatch.producerId(),
                originalBatch.producerEpoch(), originalBatch.baseSequence(), originalBatch.isTransactional(),
                originalBatch.isControlBatch(), originalBatch.partitionLeaderEpoch(), bufferOutputStream.limit());
+        builder.setDeleteHorizonMs(deleteHorizonMs);


Instead of adding a new setDeleteHorizonMs() method, would it be better to just add another constructor for MemoryRecordsBuilder that includes DeleteHorizonMs? This way, it's more consistent with how DeleteHorizonMs is set in another classes.

@junrao That wouldn't work. Checkstyle prohibits any constructor which have more than 13 parameters. Adding a new one would make it 14. I think that we should do a builder pattern instead?

We could probably just bump up the number of parameter limit in the checkstyle file to 14.

ConcurrencyPractitioner · 2019-12-08T19:32:17Z

Print statements are temporary.

ConcurrencyPractitioner · 2019-12-27T03:43:24Z

Cool, so I fixed the majority of the tests.

ConcurrencyPractitioner · 2019-12-27T04:44:37Z

Retest this please.

ConcurrencyPractitioner · 2019-12-30T18:25:45Z

@junrao Feel free to leave some comments. I fixed all the failing tests, just need to add some new ones. Will do so today.

ConcurrencyPractitioner · 2020-01-02T01:01:39Z

Retest this please.

ConcurrencyPractitioner · 2020-01-02T04:36:42Z

Migrating to #7884 .

ConcurrencyPractitioner added 5 commits October 27, 2019 10:21

[KAFKA-8522] Implementing proposal as outlined in KIP-534

f50a0f5

Adding comments

0886382

Adding mechanism to modify first timestamp

bff17b6

Cleaning things up

4d6b5a1

Fixing checkstyle

aac26a7

ConcurrencyPractitioner added 2 commits November 18, 2019 19:30

Fixing funny logic

f97478d

Fixing minor test

8250b90

junrao reviewed Dec 2, 2019

View reviewed changes

Addressing most comments

b300152

junrao reviewed Dec 2, 2019

View reviewed changes

Fixing all comments

7301fa9

junrao reviewed Dec 5, 2019

View reviewed changes

ConcurrencyPractitioner added 2 commits December 5, 2019 21:16

Addressing most comments

e209c4b

Attenpting zome fixes

50ec279

ConcurrencyPractitioner added 3 commits December 8, 2019 13:41

checkstyle

6ae485c

Fixing some tests

c18ff6d

Fixing most failed tests

7ed83bf

ConcurrencyPractitioner added 5 commits December 27, 2019 10:24

Modifying variable arg

d32bcb5

Modifying variable arg

6c1425c

Fixing stuff

8b2d2b1

Removing some info statements

649e924

Fixing wack structure

fdfa141

ConcurrencyPractitioner added 2 commits December 28, 2019 09:01

Fixing last remaining major issue

0b9c49c

Changing up stuff

1929e7e

ConcurrencyPractitioner added 4 commits December 30, 2019 11:34

Adding test

43fe252

Adding some comments and one more test

4138885

Adding fixed integration test

f43804f

Wrapping up test

edfcb7d

Putting up correct test

e07e5f8

ConcurrencyPractitioner closed this Jan 2, 2020

		boolean writeOriginalBatch = true;
		boolean shouldSetDeleteHorizon = !batch.isDeleteHorizonSet() && deleteHorizonMs != RecordBatch.NO_TIMESTAMP;

[KAFKA-8522] Implementing proposal as outlined in KIP-534 #7600

[KAFKA-8522] Implementing proposal as outlined in KIP-534 #7600

Conversation

ConcurrencyPractitioner commented Oct 27, 2019 • edited Loading

ConcurrencyPractitioner commented Nov 17, 2019

ConcurrencyPractitioner commented Nov 18, 2019

ConcurrencyPractitioner commented Nov 22, 2019

ConcurrencyPractitioner commented Nov 30, 2019

junrao 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

ConcurrencyPractitioner Dec 2, 2019 • edited Loading

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

ConcurrencyPractitioner commented Dec 3, 2019

junrao 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

ConcurrencyPractitioner commented Dec 8, 2019

ConcurrencyPractitioner commented Dec 27, 2019 • edited Loading

ConcurrencyPractitioner commented Dec 27, 2019

ConcurrencyPractitioner commented Dec 30, 2019

ConcurrencyPractitioner commented Jan 2, 2020

ConcurrencyPractitioner commented Jan 2, 2020

ConcurrencyPractitioner commented Oct 27, 2019 •

edited

Loading

ConcurrencyPractitioner Dec 2, 2019 •

edited

Loading

ConcurrencyPractitioner commented Dec 27, 2019 •

edited

Loading