kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject kafka git commit: KAFKA-5490; Cleaner should retain empty batch if needed to preserve producer last sequence
Date Wed, 28 Jun 2017 18:35:26 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk f05c70b06 -> cb0325d48


KAFKA-5490; Cleaner should retain empty batch if needed to preserve producer last sequence

Author: Jason Gustafson <jason@confluent.io>

Reviewers: Ismael Juma <ismael@juma.me.uk>, Guozhang Wang <wangguoz@gmail.com>, Jun Rao <junrao@gmail.com>

Closes #3406 from hachikuji/KAFKA-5490


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/cb0325d4
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/cb0325d4
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/cb0325d4

Branch: refs/heads/trunk
Commit: cb0325d484b957432048dd29419f0fa59c5f132d
Parents: f05c70b
Author: Jason Gustafson <jason@confluent.io>
Authored: Wed Jun 28 11:35:06 2017 -0700
Committer: Jason Gustafson <jason@confluent.io>
Committed: Wed Jun 28 11:35:13 2017 -0700

----------------------------------------------------------------------
 .../kafka/common/record/DefaultRecordBatch.java |   4 +-
 .../kafka/common/record/MemoryRecords.java      |  87 ++++++-----
 .../common/utils/ByteBufferOutputStream.java    |  21 ++-
 .../clients/consumer/internals/FetcherTest.java |   7 +-
 .../common/record/DefaultRecordBatchTest.java   |   1 +
 .../kafka/common/record/MemoryRecordsTest.java  | 105 ++++++++++++-
 core/src/main/scala/kafka/log/LogCleaner.scala  |  82 ++++++----
 .../src/main/scala/kafka/log/LogValidator.scala |   3 +
 .../scala/unit/kafka/log/LogCleanerTest.scala   | 155 +++++++++++++++++--
 .../src/test/scala/unit/kafka/log/LogTest.scala |  50 +++++-
 .../scala/unit/kafka/log/LogValidatorTest.scala |  40 ++++-
 11 files changed, 462 insertions(+), 93 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/cb0325d4/clients/src/main/java/org/apache/kafka/common/record/DefaultRecordBatch.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/DefaultRecordBatch.java b/clients/src/main/java/org/apache/kafka/common/record/DefaultRecordBatch.java
index e8eba8a..2262b33 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/DefaultRecordBatch.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/DefaultRecordBatch.java
@@ -409,8 +409,8 @@ public class DefaultRecordBatch extends AbstractRecordBatch implements MutableRe
                                         boolean isControlRecord) {
         int offsetDelta = (int) (lastOffset - baseOffset);
         writeHeader(buffer, baseOffset, offsetDelta, DefaultRecordBatch.RECORD_BATCH_OVERHEAD, magic,
-                CompressionType.NONE, timestampType, timestamp, timestamp, producerId, producerEpoch, baseSequence,
-                isTransactional, isControlRecord, partitionLeaderEpoch, 0);
+                CompressionType.NONE, timestampType, RecordBatch.NO_TIMESTAMP, timestamp, producerId,
+                producerEpoch, baseSequence, isTransactional, isControlRecord, partitionLeaderEpoch, 0);
     }
 
     static void writeHeader(ByteBuffer buffer,

http://git-wip-us.apache.org/repos/asf/kafka/blob/cb0325d4/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
index 1427421..d3f2444 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
@@ -17,6 +17,7 @@
 package org.apache.kafka.common.record;
 
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.record.MemoryRecords.RecordFilter.BatchRetention;
 import org.apache.kafka.common.utils.ByteBufferOutputStream;
 import org.apache.kafka.common.utils.CloseableIterator;
 import org.slf4j.Logger;
@@ -151,7 +152,8 @@ public class MemoryRecords extends AbstractRecords {
         for (MutableRecordBatch batch : batches) {
             bytesRead += batch.sizeInBytes();
 
-            if (filter.shouldDiscard(batch))
+            BatchRetention batchRetention = filter.checkBatchRetention(batch);
+            if (batchRetention == BatchRetention.DELETE)
                 continue;
 
             // We use the absolute offset to decide whether to retain the message or not. Due to KAFKA-4298, we have to
@@ -168,7 +170,7 @@ public class MemoryRecords extends AbstractRecords {
                     Record record = iterator.next();
                     messagesRead += 1;
 
-                    if (filter.shouldRetain(batch, record)) {
+                    if (filter.shouldRetainRecord(batch, record)) {
                         // Check for log corruption due to KAFKA-4298. If we find it, make sure that we overwrite
                         // the corrupted batch with correct data.
                         if (!record.hasMagic(batchMagic))
@@ -184,33 +186,44 @@ public class MemoryRecords extends AbstractRecords {
                 }
             }
 
-            if (writeOriginalBatch) {
-                batch.writeTo(bufferOutputStream);
-                messagesRetained += retainedRecords.size();
-                bytesRetained += batch.sizeInBytes();
-                if (batch.maxTimestamp() > maxTimestamp) {
-                    maxTimestamp = batch.maxTimestamp();
-                    shallowOffsetOfMaxTimestamp = batch.lastOffset();
-                }
-            } else if (!retainedRecords.isEmpty()) {
-                MemoryRecordsBuilder builder = buildRetainedRecordsInto(batch, retainedRecords, bufferOutputStream);
-                MemoryRecords records = builder.build();
-                int filteredBatchSize = records.sizeInBytes();
-
-                messagesRetained += retainedRecords.size();
-                bytesRetained += filteredBatchSize;
-
-                if (filteredBatchSize > batch.sizeInBytes() && filteredBatchSize > maxRecordBatchSize)
-                    log.warn("Record batch from {} with last offset {} exceeded max record batch size {} after cleaning " +
-                                    "(new size is {}). Consumers with version earlier than 0.10.1.0 may need to " +
-                                    "increase their fetch sizes.",
-                            partition, batch.lastOffset(), maxRecordBatchSize, filteredBatchSize);
-
-                MemoryRecordsBuilder.RecordsInfo info = builder.info();
-                if (info.maxTimestamp > maxTimestamp) {
-                    maxTimestamp = info.maxTimestamp;
-                    shallowOffsetOfMaxTimestamp = info.shallowOffsetOfMaxTimestamp;
+            if (!retainedRecords.isEmpty()) {
+                if (writeOriginalBatch) {
+                    batch.writeTo(bufferOutputStream);
+                    messagesRetained += retainedRecords.size();
+                    bytesRetained += batch.sizeInBytes();
+                    if (batch.maxTimestamp() > maxTimestamp) {
+                        maxTimestamp = batch.maxTimestamp();
+                        shallowOffsetOfMaxTimestamp = batch.lastOffset();
+                    }
+                } else {
+                    MemoryRecordsBuilder builder = buildRetainedRecordsInto(batch, retainedRecords, bufferOutputStream);
+                    MemoryRecords records = builder.build();
+                    int filteredBatchSize = records.sizeInBytes();
+
+                    messagesRetained += retainedRecords.size();
+                    bytesRetained += filteredBatchSize;
+
+                    if (filteredBatchSize > batch.sizeInBytes() && filteredBatchSize > maxRecordBatchSize)
+                        log.warn("Record batch from {} with last offset {} exceeded max record batch size {} after cleaning " +
+                                        "(new size is {}). Consumers with version earlier than 0.10.1.0 may need to " +
+                                        "increase their fetch sizes.",
+                                partition, batch.lastOffset(), maxRecordBatchSize, filteredBatchSize);
+
+                    MemoryRecordsBuilder.RecordsInfo info = builder.info();
+                    if (info.maxTimestamp > maxTimestamp) {
+                        maxTimestamp = info.maxTimestamp;
+                        shallowOffsetOfMaxTimestamp = info.shallowOffsetOfMaxTimestamp;
+                    }
                 }
+            } else if (batchRetention == BatchRetention.RETAIN_EMPTY) {
+                if (batchMagic < RecordBatch.MAGIC_VALUE_V2)
+                    throw new IllegalStateException("Empty batches are only supported for magic v2 and above");
+
+                bufferOutputStream.ensureRemaining(DefaultRecordBatch.RECORD_BATCH_OVERHEAD);
+                DefaultRecordBatch.writeEmptyHeader(bufferOutputStream.buffer(), batchMagic, batch.producerId(),
+                        batch.producerEpoch(), batch.baseSequence(), batch.baseOffset(), batch.lastOffset(),
+                        batch.partitionLeaderEpoch(), batch.timestampType(), batch.maxTimestamp(),
+                        batch.isTransactional(), batch.isControlBatch());
             }
 
             // If we had to allocate a new buffer to fit the filtered output (see KAFKA-5316), return early to
@@ -300,20 +313,24 @@ public class MemoryRecords extends AbstractRecords {
     }
 
     public static abstract class RecordFilter {
+        public enum BatchRetention {
+            DELETE, // Delete the batch without inspecting records
+            RETAIN_EMPTY, // Retain the batch even if it is empty
+            DELETE_EMPTY  // Delete the batch if it is empty
+        }
+
         /**
          * Check whether the full batch can be discarded (i.e. whether we even need to
          * check the records individually).
          */
-        protected boolean shouldDiscard(RecordBatch batch) {
-            return false;
-        }
+        protected abstract BatchRetention checkBatchRetention(RecordBatch batch);
 
         /**
-         * Check whether a record should be retained in the log. Only records from
-         * batches which were not discarded with {@link #shouldDiscard(RecordBatch)}
-         * will be considered.
+         * Check whether a record should be retained in the log. Note that {@link #checkBatchRetention(RecordBatch)}
+         * is used prior to checking individual record retention. Only records from batches which were not
+         * explicitly discarded with {@link BatchRetention#DELETE} will be considered.
          */
-        protected abstract boolean shouldRetain(RecordBatch recordBatch, Record record);
+        protected abstract boolean shouldRetainRecord(RecordBatch recordBatch, Record record);
     }
 
     public static class FilterResult {

http://git-wip-us.apache.org/repos/asf/kafka/blob/cb0325d4/clients/src/main/java/org/apache/kafka/common/utils/ByteBufferOutputStream.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/ByteBufferOutputStream.java b/clients/src/main/java/org/apache/kafka/common/utils/ByteBufferOutputStream.java
index 2b13e7e..43e3bba 100644
--- a/clients/src/main/java/org/apache/kafka/common/utils/ByteBufferOutputStream.java
+++ b/clients/src/main/java/org/apache/kafka/common/utils/ByteBufferOutputStream.java
@@ -62,17 +62,17 @@ public class ByteBufferOutputStream extends OutputStream {
     }
 
     public void write(int b) {
-        maybeExpandBuffer(1);
+        ensureRemaining(1);
         buffer.put((byte) b);
     }
 
     public void write(byte[] bytes, int off, int len) {
-        maybeExpandBuffer(len);
+        ensureRemaining(len);
         buffer.put(bytes, off, len);
     }
 
     public void write(ByteBuffer sourceBuffer) {
-        maybeExpandBuffer(sourceBuffer.remaining());
+        ensureRemaining(sourceBuffer.remaining());
         buffer.put(sourceBuffer);
     }
 
@@ -93,7 +93,7 @@ public class ByteBufferOutputStream extends OutputStream {
     }
 
     public void position(int position) {
-        maybeExpandBuffer(position - buffer.position());
+        ensureRemaining(position - buffer.position());
         buffer.position(position);
     }
 
@@ -105,9 +105,16 @@ public class ByteBufferOutputStream extends OutputStream {
         return initialCapacity;
     }
 
-    private void maybeExpandBuffer(int remainingRequired) {
-        if (remainingRequired > buffer.remaining())
-            expandBuffer(remainingRequired);
+    /**
+     * Ensure there is enough space to write some number of bytes, expanding the underlying buffer if necessary.
+     * This can be used to avoid incremental expansions through calls to {@link #write(int)} when you know how
+     * many total bytes are needed.
+     *
+     * @param remainingBytesRequired The number of bytes required
+     */
+    public void ensureRemaining(int remainingBytesRequired) {
+        if (remainingBytesRequired > buffer.remaining())
+            expandBuffer(remainingBytesRequired);
     }
 
     private void expandBuffer(int remainingRequired) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/cb0325d4/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
index 813a277..0801979 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
@@ -1641,7 +1641,12 @@ public class FetcherTest {
         // Remove the last record to simulate compaction
         MemoryRecords.FilterResult result = records.filterTo(tp1, new MemoryRecords.RecordFilter() {
             @Override
-            protected boolean shouldRetain(RecordBatch recordBatch, Record record) {
+            protected BatchRetention checkBatchRetention(RecordBatch batch) {
+                return BatchRetention.DELETE_EMPTY;
+            }
+
+            @Override
+            protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
                 return record.key() != null;
             }
         }, ByteBuffer.allocate(1024), Integer.MAX_VALUE, BufferSupplier.NO_CACHING);

http://git-wip-us.apache.org/repos/asf/kafka/blob/cb0325d4/clients/src/test/java/org/apache/kafka/common/record/DefaultRecordBatchTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/record/DefaultRecordBatchTest.java b/clients/src/test/java/org/apache/kafka/common/record/DefaultRecordBatchTest.java
index 6129c04..587bf14 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/DefaultRecordBatchTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/DefaultRecordBatchTest.java
@@ -63,6 +63,7 @@ public class DefaultRecordBatchTest {
                     assertEquals(isTransactional, batch.isTransactional());
                     assertEquals(timestampType, batch.timestampType());
                     assertEquals(timestamp, batch.maxTimestamp());
+                    assertEquals(RecordBatch.NO_TIMESTAMP, batch.baseTimestamp());
                     assertEquals(isControlBatch, batch.isControlBatch());
                 }
             }

http://git-wip-us.apache.org/repos/asf/kafka/blob/cb0325d4/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java b/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java
index 21f6d53..de00378 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java
@@ -18,6 +18,7 @@ package org.apache.kafka.common.record;
 
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.record.MemoryRecords.RecordFilter.BatchRetention;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.test.TestUtils;
 import org.junit.Test;
@@ -26,6 +27,7 @@ import org.junit.runners.Parameterized;
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 
@@ -239,6 +241,94 @@ public class MemoryRecordsTest {
     }
 
     @Test
+    public void testFilterToEmptyBatchRetention() {
+        if (magic >= RecordBatch.MAGIC_VALUE_V2) {
+            for (boolean isTransactional : Arrays.asList(true, false)) {
+                ByteBuffer buffer = ByteBuffer.allocate(2048);
+                long producerId = 23L;
+                short producerEpoch = 5;
+                long baseOffset = 3L;
+                int baseSequence = 10;
+                int partitionLeaderEpoch = 293;
+
+                MemoryRecordsBuilder builder = MemoryRecords.builder(buffer, magic, compression, TimestampType.CREATE_TIME,
+                        baseOffset, RecordBatch.NO_TIMESTAMP, producerId, producerEpoch, baseSequence, isTransactional,
+                        partitionLeaderEpoch);
+                builder.append(11L, "2".getBytes(), "b".getBytes());
+                builder.append(12L, "3".getBytes(), "c".getBytes());
+                builder.close();
+
+                ByteBuffer filtered = ByteBuffer.allocate(2048);
+                builder.build().filterTo(new TopicPartition("foo", 0), new MemoryRecords.RecordFilter() {
+                    @Override
+                    protected BatchRetention checkBatchRetention(RecordBatch batch) {
+                        // retain all batches
+                        return BatchRetention.RETAIN_EMPTY;
+                    }
+
+                    @Override
+                    protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
+                        // delete the records
+                        return false;
+                    }
+                }, filtered, Integer.MAX_VALUE, BufferSupplier.NO_CACHING);
+
+                filtered.flip();
+                MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered);
+
+                List<MutableRecordBatch> batches = TestUtils.toList(filteredRecords.batches());
+                assertEquals(1, batches.size());
+
+                MutableRecordBatch batch = batches.get(0);
+                assertEquals(0, batch.countOrNull().intValue());
+                assertEquals(12L, batch.maxTimestamp());
+                assertEquals(TimestampType.CREATE_TIME, batch.timestampType());
+                assertEquals(baseOffset, batch.baseOffset());
+                assertEquals(baseOffset + 1, batch.lastOffset());
+                assertEquals(baseSequence, batch.baseSequence());
+                assertEquals(baseSequence + 1, batch.lastSequence());
+                assertEquals(isTransactional, batch.isTransactional());
+            }
+        }
+    }
+
+    @Test
+    public void testEmptyBatchDeletion() {
+        if (magic >= RecordBatch.MAGIC_VALUE_V2) {
+            for (final BatchRetention deleteRetention : Arrays.asList(BatchRetention.DELETE, BatchRetention.DELETE_EMPTY)) {
+                ByteBuffer buffer = ByteBuffer.allocate(DefaultRecordBatch.RECORD_BATCH_OVERHEAD);
+                long producerId = 23L;
+                short producerEpoch = 5;
+                long baseOffset = 3L;
+                int baseSequence = 10;
+                int partitionLeaderEpoch = 293;
+
+                DefaultRecordBatch.writeEmptyHeader(buffer, RecordBatch.MAGIC_VALUE_V2, producerId, producerEpoch,
+                        baseSequence, baseOffset, baseOffset, partitionLeaderEpoch, TimestampType.CREATE_TIME,
+                        System.currentTimeMillis(), false, false);
+                buffer.flip();
+
+                ByteBuffer filtered = ByteBuffer.allocate(2048);
+                MemoryRecords.readableRecords(buffer).filterTo(new TopicPartition("foo", 0), new MemoryRecords.RecordFilter() {
+                    @Override
+                    protected BatchRetention checkBatchRetention(RecordBatch batch) {
+                        return deleteRetention;
+                    }
+
+                    @Override
+                    protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
+                        return false;
+                    }
+                }, filtered, Integer.MAX_VALUE, BufferSupplier.NO_CACHING);
+
+                filtered.flip();
+                MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered);
+                assertEquals(0, filteredRecords.sizeInBytes());
+            }
+        }
+    }
+
+    @Test
     public void testBuildEndTxnMarker() {
         if (magic >= RecordBatch.MAGIC_VALUE_V2) {
             long producerId = 73;
@@ -303,13 +393,15 @@ public class MemoryRecordsTest {
             ByteBuffer filtered = ByteBuffer.allocate(2048);
             MemoryRecords.readableRecords(buffer).filterTo(new TopicPartition("foo", 0), new MemoryRecords.RecordFilter() {
                 @Override
-                protected boolean shouldDiscard(RecordBatch batch) {
+                protected BatchRetention checkBatchRetention(RecordBatch batch) {
                     // discard the second and fourth batches
-                    return batch.lastOffset() == 2L || batch.lastOffset() == 6L;
+                    if (batch.lastOffset() == 2L || batch.lastOffset() == 6L)
+                        return BatchRetention.DELETE;
+                    return BatchRetention.DELETE_EMPTY;
                 }
 
                 @Override
-                protected boolean shouldRetain(RecordBatch recordBatch, Record record) {
+                protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
                     return true;
                 }
             }, filtered, Integer.MAX_VALUE, BufferSupplier.NO_CACHING);
@@ -683,7 +775,12 @@ public class MemoryRecordsTest {
 
     private static class RetainNonNullKeysFilter extends MemoryRecords.RecordFilter {
         @Override
-        public boolean shouldRetain(RecordBatch batch, Record record) {
+        protected BatchRetention checkBatchRetention(RecordBatch batch) {
+            return BatchRetention.DELETE_EMPTY;
+        }
+
+        @Override
+        public boolean shouldRetainRecord(RecordBatch batch, Record record) {
             return record.hasKey();
         }
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/cb0325d4/core/src/main/scala/kafka/log/LogCleaner.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/LogCleaner.scala b/core/src/main/scala/kafka/log/LogCleaner.scala
index 5222487..d8a86db 100644
--- a/core/src/main/scala/kafka/log/LogCleaner.scala
+++ b/core/src/main/scala/kafka/log/LogCleaner.scala
@@ -30,6 +30,7 @@ import org.apache.kafka.common.record._
 import org.apache.kafka.common.utils.Time
 import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.record.MemoryRecords.RecordFilter
+import org.apache.kafka.common.record.MemoryRecords.RecordFilter.BatchRetention
 
 import scala.collection.mutable
 import scala.collection.JavaConverters._
@@ -63,7 +64,22 @@ import scala.collection.JavaConverters._
  * The cleaner will only retain delete records for a period of time to avoid accumulating space indefinitely. This period of time is configurable on a per-topic
  * basis and is measured from the time the segment enters the clean portion of the log (at which point any prior message with that key has been removed).
  * Delete markers in the clean section of the log that are older than this time will not be retained when log segments are being recopied as part of cleaning.
- * 
+ *
+ * Note that cleaning is more complicated with the idempotent/transactional producer capabilities. The following
+ * are the key points:
+ *
+ * 1. In order to maintain sequence number continuity for active producers, we always retain the last batch
+ *    from each producerId, even if all the records from the batch have been removed. The batch will be removed
+ *    once the producer either writes a new batch or is expired due to inactivity.
+ * 2. We do not clean beyond the last stable offset. This ensures that all records observed by the cleaner have
+ *    been decided (i.e. committed or aborted). In particular, this allows us to use the transaction index to
+ *    collect the aborted transactions ahead of time.
+ * 3. Records from aborted transactions are removed by the cleaner immediately without regard to record keys.
+ * 4. Transaction markers are retained until all record batches from the same transaction have been removed and
+ *    a sufficient amount of time has passed to reasonably ensure that an active consumer wouldn't consume any
+ *    data from the transaction prior to reaching the offset of the marker. This follows the same logic used for
+ *    tombstone deletion.
+ *
  * @param config Configuration parameters for the cleaner
  * @param logDirs The directories where offset checkpoints reside
  * @param logs The pool of logs
@@ -478,34 +494,29 @@ private[log] class Cleaner(val id: Int,
                              activeProducers: Map[Long, ProducerIdEntry],
                              stats: CleanerStats) {
     val logCleanerFilter = new RecordFilter {
-      var retainLastBatchSequence: Boolean = false
-      var discardBatchRecords: Boolean = false
+      var discardBatchRecords: Boolean = _
 
-      override def shouldDiscard(batch: RecordBatch): Boolean = {
+      override def checkBatchRetention(batch: RecordBatch): BatchRetention = {
         // we piggy-back on the tombstone retention logic to delay deletion of transaction markers.
         // note that we will never delete a marker until all the records from that transaction are removed.
         discardBatchRecords = shouldDiscardBatch(batch, transactionMetadata, retainTxnMarkers = retainDeletes)
 
         // check if the batch contains the last sequence number for the producer. if so, we cannot
         // remove the batch just yet or the producer may see an out of sequence error.
-        if (batch.hasProducerId && activeProducers.get(batch.producerId).exists(_.lastSeq == batch.lastSequence)) {
-          retainLastBatchSequence = true
-          false
-        } else {
-          retainLastBatchSequence = false
-          discardBatchRecords
-        }
+        if (batch.hasProducerId && activeProducers.get(batch.producerId).exists(_.lastSeq == batch.lastSequence))
+          BatchRetention.RETAIN_EMPTY
+        else if (discardBatchRecords)
+          BatchRetention.DELETE
+        else
+          BatchRetention.DELETE_EMPTY
       }
 
-      override def shouldRetain(batch: RecordBatch, record: Record): Boolean = {
-        if (retainLastBatchSequence && batch.lastSequence == record.sequence)
-          // always retain the record with the last sequence number
-          true
-        else if (discardBatchRecords)
-          // remove the record if the batch would have otherwise been discarded
+      override def shouldRetainRecord(batch: RecordBatch, record: Record): Boolean = {
+        if (discardBatchRecords)
+          // The batch is only retained to preserve producer sequence information; the records can be removed
           false
         else
-          shouldRetainRecord(source, map, retainDeletes, batch, record, stats)
+          Cleaner.this.shouldRetainRecord(source, map, retainDeletes, batch, record, stats)
       }
     }
 
@@ -731,7 +742,8 @@ private[log] class Cleaner(val id: Int,
         } else {
           val isAborted = transactionMetadata.onBatchRead(batch)
           if (isAborted) {
-            // abort markers are supported in v2 and above, which means count is defined
+            // If the batch is aborted, do not bother populating the offset map.
+            // Note that abort markers are supported in v2 and above, which means count is defined.
             stats.indexMessagesRead(batch.countOrNull)
           } else {
             for (record <- batch.asScala) {
@@ -849,7 +861,7 @@ private[log] object CleanedTransactionMetadata {
 private[log] class CleanedTransactionMetadata(val abortedTransactions: mutable.PriorityQueue[AbortedTxn],
                                               val transactionIndex: Option[TransactionIndex] = None) {
   val ongoingCommittedTxns = mutable.Set.empty[Long]
-  val ongoingAbortedTxns = mutable.Map.empty[Long, AbortedTxn]
+  val ongoingAbortedTxns = mutable.Map.empty[Long, AbortedTransactionMetadata]
 
   /**
    * Update the cleaned transaction state with a control batch that has just been traversed by the cleaner.
@@ -863,14 +875,16 @@ private[log] class CleanedTransactionMetadata(val abortedTransactions: mutable.P
     val producerId = controlBatch.producerId
     controlType match {
       case ControlRecordType.ABORT =>
-        val maybeAbortedTxn = ongoingAbortedTxns.remove(producerId)
-        maybeAbortedTxn.foreach { abortedTxn =>
-          transactionIndex.foreach(_.append(abortedTxn))
+        ongoingAbortedTxns.remove(producerId) match {
+          // Retain the marker until all batches from the transaction have been removed
+          case Some(abortedTxnMetadata) if abortedTxnMetadata.lastObservedBatchOffset.isDefined =>
+            transactionIndex.foreach(_.append(abortedTxnMetadata.abortedTxn))
+            false
+          case _ => true
         }
-        true
 
       case ControlRecordType.COMMIT =>
-        // this marker is eligible for deletion if we didn't traverse any records from the transaction
+        // This marker is eligible for deletion if we didn't traverse any batches from the transaction
         !ongoingCommittedTxns.remove(producerId)
 
       case _ => false
@@ -880,7 +894,7 @@ private[log] class CleanedTransactionMetadata(val abortedTransactions: mutable.P
   private def consumeAbortedTxnsUpTo(offset: Long): Unit = {
     while (abortedTransactions.headOption.exists(_.firstOffset <= offset)) {
       val abortedTxn = abortedTransactions.dequeue()
-      ongoingAbortedTxns += abortedTxn.producerId -> abortedTxn
+      ongoingAbortedTxns += abortedTxn.producerId -> new AbortedTransactionMetadata(abortedTxn)
     }
   }
 
@@ -891,11 +905,13 @@ private[log] class CleanedTransactionMetadata(val abortedTransactions: mutable.P
   def onBatchRead(batch: RecordBatch): Boolean = {
     consumeAbortedTxnsUpTo(batch.lastOffset)
     if (batch.isTransactional) {
-      if (ongoingAbortedTxns.contains(batch.producerId))
-        true
-      else {
-        ongoingCommittedTxns += batch.producerId
-        false
+      ongoingAbortedTxns.get(batch.producerId) match {
+        case Some(abortedTransactionMetadata) =>
+          abortedTransactionMetadata.lastObservedBatchOffset = Some(batch.lastOffset)
+          true
+        case None =>
+          ongoingCommittedTxns += batch.producerId
+          false
       }
     } else {
       false
@@ -903,3 +919,7 @@ private[log] class CleanedTransactionMetadata(val abortedTransactions: mutable.P
   }
 
 }
+
+private class AbortedTransactionMetadata(val abortedTxn: AbortedTxn) {
+  var lastObservedBatchOffset: Option[Long] = None
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/cb0325d4/core/src/main/scala/kafka/log/LogValidator.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/LogValidator.scala b/core/src/main/scala/kafka/log/LogValidator.scala
index e7d7963..b6b20e3 100644
--- a/core/src/main/scala/kafka/log/LogValidator.scala
+++ b/core/src/main/scala/kafka/log/LogValidator.scala
@@ -78,6 +78,9 @@ private[kafka] object LogValidator extends Logging {
 
       if (batch.isControlBatch)
         throw new InvalidRecordException("Clients are not allowed to write control records")
+
+      if (Option(batch.countOrNull).contains(0))
+        throw new InvalidRecordException("Record batches must contain at least one record")
     }
 
     if (batch.isTransactional && toMagic < RecordBatch.MAGIC_VALUE_V2)

http://git-wip-us.apache.org/repos/asf/kafka/blob/cb0325d4/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
index dabd2d6..19ea699 100755
--- a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
@@ -19,7 +19,7 @@ package kafka.log
 
 import java.io.File
 import java.nio._
-import java.nio.file.{Files, Paths}
+import java.nio.file.Paths
 import java.util.Properties
 
 import kafka.common._
@@ -107,8 +107,10 @@ class LogCleanerTest extends JUnitSuite {
 
     log.roll()
     cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0L, log.activeSegment.baseOffset))
-    assertEquals(List(2, 3, 3, 4, 1, 4), keysInLog(log))
-    assertEquals(List(1, 2, 3, 5, 6, 7), offsetsInLog(log))
+    assertEquals(List(2, 5, 7), lastOffsetsPerBatchInLog(log))
+    assertEquals(Map(pid1 -> 2, pid2 -> 2, pid3 -> 1), lastSequencesInLog(log))
+    assertEquals(List(2, 3, 1, 4), keysInLog(log))
+    assertEquals(List(1, 3, 6, 7), offsetsInLog(log))
 
     // we have to reload the log to validate that the cleaner maintained sequence numbers correctly
     def reloadLog(): Unit = {
@@ -137,8 +139,10 @@ class LogCleanerTest extends JUnitSuite {
     appendIdempotentAsLeader(log, pid4, producerEpoch)(Seq(2))
     log.roll()
     cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0L, log.activeSegment.baseOffset))
-    assertEquals(List(3, 3, 4, 1, 4, 2), keysInLog(log))
-    assertEquals(List(2, 3, 5, 6, 7, 8), offsetsInLog(log))
+    assertEquals(Map(pid1 -> 2, pid2 -> 2, pid3 -> 1, pid4 -> 0), lastSequencesInLog(log))
+    assertEquals(List(2, 5, 7, 8), lastOffsetsPerBatchInLog(log))
+    assertEquals(List(3, 1, 4, 2), keysInLog(log))
+    assertEquals(List(3, 6, 7, 8), offsetsInLog(log))
 
     reloadLog()
 
@@ -266,12 +270,59 @@ class LogCleanerTest extends JUnitSuite {
     assertEquals(List(3, 4, 5, 6, 7, 8), offsetsInLog(log))
 
     // clean again with large delete horizon and verify the marker is removed
-    cleaner.doClean(LogToClean(tp, log, dirtyOffset, 100L), deleteHorizonMs = Long.MaxValue)
+    dirtyOffset = cleaner.doClean(LogToClean(tp, log, dirtyOffset, 100L), deleteHorizonMs = Long.MaxValue)._1
     assertEquals(List(2, 1, 3), keysInLog(log))
     assertEquals(List(4, 5, 6, 7, 8), offsetsInLog(log))
   }
 
   @Test
+  def testCommitMarkerRetentionWithEmptyBatch(): Unit = {
+    val tp = new TopicPartition("test", 0)
+    val cleaner = makeCleaner(Int.MaxValue)
+    val logProps = new Properties()
+    logProps.put(LogConfig.SegmentBytesProp, 256: java.lang.Integer)
+    val log = makeLog(config = LogConfig.fromProps(logConfig.originals, logProps))
+
+    val producerEpoch = 0.toShort
+    val producerId = 1L
+    val appendProducer = appendTransactionalAsLeader(log, producerId, producerEpoch)
+
+    appendProducer(Seq(2, 3)) // batch last offset is 1
+    log.appendAsLeader(commitMarker(producerId, producerEpoch), leaderEpoch = 0, isFromClient = false)
+    log.roll()
+
+    log.appendAsLeader(record(2, 2), leaderEpoch = 0)
+    log.appendAsLeader(record(3, 3), leaderEpoch = 0)
+    log.roll()
+
+    // first time through the records are removed
+    var dirtyOffset = cleaner.doClean(LogToClean(tp, log, 0L, 100L), deleteHorizonMs = Long.MaxValue)._1
+    assertEquals(List(2, 3), keysInLog(log))
+    assertEquals(List(2, 3, 4), offsetsInLog(log)) // commit marker is retained
+    assertEquals(List(1, 2, 3, 4), lastOffsetsPerBatchInLog(log)) // empty batch is retained
+
+    // the empty batch remains if cleaned again because it still holds the last sequence
+    dirtyOffset = cleaner.doClean(LogToClean(tp, log, dirtyOffset, 100L), deleteHorizonMs = Long.MaxValue)._1
+    assertEquals(List(2, 3), keysInLog(log))
+    assertEquals(List(2, 3, 4), offsetsInLog(log)) // commit marker is still retained
+    assertEquals(List(1, 2, 3, 4), lastOffsetsPerBatchInLog(log)) // empty batch is retained
+
+    // append a new record from the producer to allow cleaning of the empty batch
+    appendProducer(Seq(1))
+    log.roll()
+
+    dirtyOffset = cleaner.doClean(LogToClean(tp, log, dirtyOffset, 100L), deleteHorizonMs = Long.MaxValue)._1
+    assertEquals(List(2, 3, 1), keysInLog(log))
+    assertEquals(List(2, 3, 4, 5), offsetsInLog(log)) // commit marker is still retained
+    assertEquals(List(2, 3, 4, 5), lastOffsetsPerBatchInLog(log)) // empty batch should be gone
+
+    dirtyOffset = cleaner.doClean(LogToClean(tp, log, dirtyOffset, 100L), deleteHorizonMs = Long.MaxValue)._1
+    assertEquals(List(2, 3, 1), keysInLog(log))
+    assertEquals(List(3, 4, 5), offsetsInLog(log)) // commit marker is gone
+    assertEquals(List(3, 4, 5), lastOffsetsPerBatchInLog(log)) // empty batch is gone
+  }
+
+  @Test
   def testAbortMarkerRemoval(): Unit = {
     val tp = new TopicPartition("test", 0)
     val cleaner = makeCleaner(Int.MaxValue)
@@ -301,6 +352,65 @@ class LogCleanerTest extends JUnitSuite {
     assertEquals(List(4, 5), offsetsInLog(log))
   }
 
+  @Test
+  def testAbortMarkerRetentionWithEmptyBatch(): Unit = {
+    val tp = new TopicPartition("test", 0)
+    val cleaner = makeCleaner(Int.MaxValue)
+    val logProps = new Properties()
+    logProps.put(LogConfig.SegmentBytesProp, 256: java.lang.Integer)
+    val log = makeLog(config = LogConfig.fromProps(logConfig.originals, logProps))
+
+    val producerEpoch = 0.toShort
+    val producerId = 1L
+    val appendProducer = appendTransactionalAsLeader(log, producerId, producerEpoch)
+
+    appendProducer(Seq(2, 3)) // batch last offset is 1
+    log.appendAsLeader(abortMarker(producerId, producerEpoch), leaderEpoch = 0, isFromClient = false)
+    log.roll()
+
+    def assertAbortedTransactionIndexed(): Unit = {
+      val abortedTxns = log.collectAbortedTransactions(0L, 100L)
+      assertEquals(1, abortedTxns.size)
+      assertEquals(producerId, abortedTxns.head.producerId)
+      assertEquals(0, abortedTxns.head.firstOffset)
+      assertEquals(2, abortedTxns.head.lastOffset)
+    }
+
+    assertAbortedTransactionIndexed()
+
+    // first time through the records are removed
+    var dirtyOffset = cleaner.doClean(LogToClean(tp, log, 0L, 100L), deleteHorizonMs = Long.MaxValue)._1
+    assertAbortedTransactionIndexed()
+    assertEquals(List(), keysInLog(log))
+    assertEquals(List(2), offsetsInLog(log)) // abort marker is retained
+    assertEquals(List(1, 2), lastOffsetsPerBatchInLog(log)) // empty batch is retained
+
+    // the empty batch remains if cleaned again because it still holds the last sequence
+    dirtyOffset = cleaner.doClean(LogToClean(tp, log, dirtyOffset, 100L), deleteHorizonMs = Long.MaxValue)._1
+    assertAbortedTransactionIndexed()
+    assertEquals(List(), keysInLog(log))
+    assertEquals(List(2), offsetsInLog(log)) // abort marker is still retained
+    assertEquals(List(1, 2), lastOffsetsPerBatchInLog(log)) // empty batch is retained
+
+    // now update the last sequence so that the empty batch can be removed
+    appendProducer(Seq(1))
+    log.roll()
+
+    dirtyOffset = cleaner.doClean(LogToClean(tp, log, dirtyOffset, 100L), deleteHorizonMs = Long.MaxValue)._1
+    assertAbortedTransactionIndexed()
+    assertEquals(List(1), keysInLog(log))
+    assertEquals(List(2, 3), offsetsInLog(log)) // abort marker is not yet gone because we read the empty batch
+    assertEquals(List(2, 3), lastOffsetsPerBatchInLog(log)) // but we do not preserve the empty batch
+
+    dirtyOffset = cleaner.doClean(LogToClean(tp, log, dirtyOffset, 100L), deleteHorizonMs = Long.MaxValue)._1
+    assertEquals(List(1), keysInLog(log))
+    assertEquals(List(3), offsetsInLog(log)) // abort marker is gone
+    assertEquals(List(3), lastOffsetsPerBatchInLog(log))
+
+    // we do not bother retaining the aborted transaction in the index
+    assertEquals(0, log.collectAbortedTransactions(0L, 100L).size)
+  }
+
   /**
    * Test log cleaning with logs containing messages larger than default message size
    */
@@ -404,8 +514,10 @@ class LogCleanerTest extends JUnitSuite {
     log.roll()
 
     cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0L, log.activeSegment.baseOffset))
-    assertEquals(List(0, 0, 1), keysInLog(log))
-    assertEquals(List(1, 3, 4), offsetsInLog(log))
+    assertEquals(List(1, 3, 4), lastOffsetsPerBatchInLog(log))
+    assertEquals(Map(1L -> 0, 2L -> 1, 3L -> 0), lastSequencesInLog(log))
+    assertEquals(List(0, 1), keysInLog(log))
+    assertEquals(List(3, 4), offsetsInLog(log))
   }
 
   @Test
@@ -425,8 +537,20 @@ class LogCleanerTest extends JUnitSuite {
     log.roll()
 
     cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0L, log.activeSegment.baseOffset))
-    assertEquals(List(3), keysInLog(log))
-    assertEquals(List(2, 3), offsetsInLog(log))
+    assertEquals(List(2, 3), lastOffsetsPerBatchInLog(log))
+    assertEquals(Map(producerId -> 2), lastSequencesInLog(log))
+    assertEquals(List(), keysInLog(log))
+    assertEquals(List(3), offsetsInLog(log))
+
+    // Append a new entry from the producer and verify that the empty batch is cleaned up
+    appendProducer(Seq(1, 5))
+    log.roll()
+    cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0L, log.activeSegment.baseOffset))
+
+    assertEquals(List(3, 5), lastOffsetsPerBatchInLog(log))
+    assertEquals(Map(producerId -> 4), lastSequencesInLog(log))
+    assertEquals(List(1, 5), keysInLog(log))
+    assertEquals(List(3, 4, 5), offsetsInLog(log))
   }
 
   @Test
@@ -588,6 +712,17 @@ class LogCleanerTest extends JUnitSuite {
       yield TestUtils.readString(record.key).toInt
   }
 
+  def lastOffsetsPerBatchInLog(log: Log): Iterable[Long] = {
+    for (segment <- log.logSegments; batch <- segment.log.batches.asScala)
+      yield batch.lastOffset
+  }
+
+  def lastSequencesInLog(log: Log): Map[Long, Int] = {
+    (for (segment <- log.logSegments;
+          batch <- segment.log.batches.asScala if !batch.isControlBatch && batch.hasProducerId)
+      yield batch.producerId -> batch.lastSequence).toMap
+  }
+
   /* extract all the offsets from a log */
   def offsetsInLog(log: Log): Iterable[Long] =
     log.logSegments.flatMap(s => s.log.records.asScala.filter(_.hasValue).filter(_.hasKey).map(m => m.offset))

http://git-wip-us.apache.org/repos/asf/kafka/blob/cb0325d4/core/src/test/scala/unit/kafka/log/LogTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/log/LogTest.scala b/core/src/test/scala/unit/kafka/log/LogTest.scala
index 630cfcf..65a4eeb 100755
--- a/core/src/test/scala/unit/kafka/log/LogTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogTest.scala
@@ -32,6 +32,7 @@ import kafka.server.{BrokerTopicStats, KafkaConfig}
 import kafka.server.epoch.{EpochEntry, LeaderEpochFileCache}
 import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.record.MemoryRecords.RecordFilter
+import org.apache.kafka.common.record.MemoryRecords.RecordFilter.BatchRetention
 import org.apache.kafka.common.record._
 import org.apache.kafka.common.requests.FetchResponse.AbortedTransaction
 import org.apache.kafka.common.requests.IsolationLevel
@@ -401,7 +402,8 @@ class LogTest {
 
     val filtered = ByteBuffer.allocate(2048)
     records.filterTo(new TopicPartition("foo", 0), new RecordFilter {
-      override def shouldRetain(recordBatch: RecordBatch, record: Record): Boolean = !record.hasKey
+      override def checkBatchRetention(batch: RecordBatch): BatchRetention = RecordFilter.BatchRetention.DELETE_EMPTY
+      override def shouldRetainRecord(recordBatch: RecordBatch, record: Record): Boolean = !record.hasKey
     }, filtered, Int.MaxValue, BufferSupplier.NO_CACHING)
     filtered.flip()
     val filteredRecords = MemoryRecords.readableRecords(filtered)
@@ -428,6 +430,49 @@ class LogTest {
   }
 
   @Test
+  def testRebuildProducerStateWithEmptyCompactedBatch() {
+    val log = createLog(2048)
+    val pid = 1L
+    val epoch = 0.toShort
+    val seq = 0
+    val baseOffset = 23L
+
+    // create an empty batch
+    val records = TestUtils.records(producerId = pid, producerEpoch = epoch, sequence = seq, baseOffset = baseOffset, records = List(
+      new SimpleRecord(System.currentTimeMillis(), "key".getBytes, "a".getBytes),
+      new SimpleRecord(System.currentTimeMillis(), "key".getBytes, "b".getBytes)))
+    records.batches.asScala.foreach(_.setPartitionLeaderEpoch(0))
+
+    val filtered = ByteBuffer.allocate(2048)
+    records.filterTo(new TopicPartition("foo", 0), new RecordFilter {
+      override def checkBatchRetention(batch: RecordBatch): BatchRetention = RecordFilter.BatchRetention.RETAIN_EMPTY
+      override def shouldRetainRecord(recordBatch: RecordBatch, record: Record): Boolean = false
+    }, filtered, Int.MaxValue, BufferSupplier.NO_CACHING)
+    filtered.flip()
+    val filteredRecords = MemoryRecords.readableRecords(filtered)
+
+    log.appendAsFollower(filteredRecords)
+
+    // append some more data and then truncate to force rebuilding of the PID map
+    val moreRecords = TestUtils.records(baseOffset = baseOffset + 2, records = List(
+      new SimpleRecord(System.currentTimeMillis(), "e".getBytes),
+      new SimpleRecord(System.currentTimeMillis(), "f".getBytes)))
+    moreRecords.batches.asScala.foreach(_.setPartitionLeaderEpoch(0))
+    log.appendAsFollower(moreRecords)
+
+    log.truncateTo(baseOffset + 2)
+
+    val activeProducers = log.activeProducers
+    assertTrue(activeProducers.contains(pid))
+
+    val entry = activeProducers(pid)
+    assertEquals(0, entry.firstSeq)
+    assertEquals(baseOffset, entry.firstOffset)
+    assertEquals(1, entry.lastSeq)
+    assertEquals(baseOffset + 1, entry.lastOffset)
+  }
+
+  @Test
   def testUpdatePidMapWithCompactedData() {
     val log = createLog(2048)
     val pid = 1L
@@ -445,7 +490,8 @@ class LogTest {
 
     val filtered = ByteBuffer.allocate(2048)
     records.filterTo(new TopicPartition("foo", 0), new RecordFilter {
-      override def shouldRetain(recordBatch: RecordBatch, record: Record): Boolean = !record.hasKey
+      override def checkBatchRetention(batch: RecordBatch): BatchRetention = RecordFilter.BatchRetention.DELETE_EMPTY
+      override def shouldRetainRecord(recordBatch: RecordBatch, record: Record): Boolean = !record.hasKey
     }, filtered, Int.MaxValue, BufferSupplier.NO_CACHING)
     filtered.flip()
     val filteredRecords = MemoryRecords.readableRecords(filtered)

http://git-wip-us.apache.org/repos/asf/kafka/blob/cb0325d4/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala b/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala
index 3ab9732..2225ca6 100644
--- a/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala
@@ -19,7 +19,7 @@ package kafka.log
 import java.nio.ByteBuffer
 
 import kafka.common.LongRef
-import kafka.message.{DefaultCompressionCodec, GZIPCompressionCodec, NoCompressionCodec, SnappyCompressionCodec}
+import kafka.message.{CompressionCodec, DefaultCompressionCodec, GZIPCompressionCodec, NoCompressionCodec, SnappyCompressionCodec}
 import org.apache.kafka.common.errors.{InvalidTimestampException, UnsupportedForMessageFormatException}
 import org.apache.kafka.common.record._
 import org.apache.kafka.test.TestUtils
@@ -909,6 +909,44 @@ class LogValidatorTest {
       isFromClient = true)
   }
 
+  @Test(expected = classOf[InvalidRecordException])
+  def testCompressedBatchWithoutRecordsNotAllowed(): Unit = {
+    testBatchWithoutRecordsNotAllowed(DefaultCompressionCodec, DefaultCompressionCodec)
+  }
+
+  @Test(expected = classOf[InvalidRecordException])
+  def testUncompressedBatchWithoutRecordsNotAllowed(): Unit = {
+    testBatchWithoutRecordsNotAllowed(NoCompressionCodec, NoCompressionCodec)
+  }
+
+  @Test(expected = classOf[InvalidRecordException])
+  def testRecompressedBatchWithoutRecordsNotAllowed(): Unit = {
+    testBatchWithoutRecordsNotAllowed(NoCompressionCodec, DefaultCompressionCodec)
+  }
+
+  private def testBatchWithoutRecordsNotAllowed(sourceCodec: CompressionCodec, targetCodec: CompressionCodec): Unit = {
+    val offset = 1234567
+    val (producerId, producerEpoch, baseSequence, isTransactional, partitionLeaderEpoch) =
+      (1324L, 10.toShort, 984, true, 40)
+    val buffer = ByteBuffer.allocate(DefaultRecordBatch.RECORD_BATCH_OVERHEAD)
+    DefaultRecordBatch.writeEmptyHeader(buffer, RecordBatch.CURRENT_MAGIC_VALUE, producerId, producerEpoch,
+      baseSequence, 0L, 5L, partitionLeaderEpoch, TimestampType.CREATE_TIME, System.currentTimeMillis(),
+      isTransactional, false)
+    buffer.flip()
+    val records = MemoryRecords.readableRecords(buffer)
+    LogValidator.validateMessagesAndAssignOffsets(records,
+      offsetCounter = new LongRef(offset),
+      now = System.currentTimeMillis(),
+      sourceCodec = sourceCodec,
+      targetCodec = targetCodec,
+      compactedTopic = false,
+      magic = RecordBatch.CURRENT_MAGIC_VALUE,
+      timestampType = TimestampType.CREATE_TIME,
+      timestampDiffMaxMs = 5000L,
+      partitionLeaderEpoch = RecordBatch.NO_PARTITION_LEADER_EPOCH,
+      isFromClient = true)
+  }
+
   private def createRecords(magicValue: Byte = RecordBatch.CURRENT_MAGIC_VALUE,
                             timestamp: Long = RecordBatch.NO_TIMESTAMP,
                             codec: CompressionType = CompressionType.NONE): MemoryRecords = {


Mime
View raw message