[KAFKA-8522] Streamline tombstone and transaction marker removal

checkstyle/checkstyle.xml

@@ -107,7 +107,7 @@
     <module name="MethodLength"/>
     <module name="ParameterNumber">
       <!-- default is 8 -->
-      <property name="max" value="13"/>
+      <property name="max" value="14"/>
     </module>
     <module name="ClassDataAbstractionCoupling">
       <!-- default is 7 -->
@@ -124,7 +124,7 @@
     </module>
     <module name="CyclomaticComplexity">
       <!-- default is 10-->
-      <property name="max" value="16"/>
+      <property name="max" value="18"/>
     </module>
     <module name="JavaNCSS">
       <!-- default is 50 -->

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

@@ -217,6 +217,16 @@ public boolean isControlBatch() {
         return false;
     }
 
+    @Override
+    public long deleteHorizonMs() {
+        return RecordBatch.NO_TIMESTAMP;
+    }
+
+    @Override
+    public boolean hasDeleteHorizonMs() {
+        return false;
+    }
+
     /**
      * Get an iterator for the nested entries contained within this batch. Note that
      * if the batch is not compressed, then this method will return an iterator over the
@@ -468,6 +478,16 @@ public long offset() {
             return buffer.getLong(OFFSET_OFFSET);
         }
 
+        @Override
+        public long deleteHorizonMs() {
+            return RecordBatch.NO_TIMESTAMP;
+        }
+
+        @Override
+        public boolean hasDeleteHorizonMs() {
+            return false;
+        }
+
         @Override
         public LegacyRecord outerRecord() {
             return record;
@@ -557,6 +577,16 @@ public long baseOffset() {
             return loadFullBatch().baseOffset();
         }
 
+        @Override
+        public long deleteHorizonMs() {
+            return RecordBatch.NO_TIMESTAMP;
+        }
+
+        @Override
+        public boolean hasDeleteHorizonMs() {
+            return false;
+        }
+
         @Override
         public long lastOffset() {
             return offset;

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

@@ -89,11 +89,15 @@
  * by the broker and is preserved after compaction. Additionally, the MaxTimestamp of an empty batch always retains
  * the previous value prior to becoming empty.
  *
+ * The delete horizon flag for the sixth bit is used to determine if the first timestamp of the batch had been set to
+ * the time for which tombstones / transaction markers needs to be removed. If it is true, then the first timestamp is
+ * the delete horizon, otherwise, it is merely the first timestamp of the record batch.
+ *
  * The current attributes are given below:
  *
- *  -------------------------------------------------------------------------------------------------
- *  | Unused (6-15) | Control (5) | Transactional (4) | Timestamp Type (3) | Compression Type (0-2) |
- *  -------------------------------------------------------------------------------------------------
+ *  ---------------------------------------------------------------------------------------------------------------------------
+ *  | Unused (7-15) | Delete Horizon Flag (6) | Control (5) | Transactional (4) | Timestamp Type (3) | Compression Type (0-2) |
+ *  ---------------------------------------------------------------------------------------------------------------------------
  */
 public class DefaultRecordBatch extends AbstractRecordBatch implements MutableRecordBatch {
     static final int BASE_OFFSET_OFFSET = 0;
@@ -128,6 +132,7 @@ public class DefaultRecordBatch extends AbstractRecordBatch implements MutableRe
     private static final byte COMPRESSION_CODEC_MASK = 0x07;
     private static final byte TRANSACTIONAL_FLAG_MASK = 0x10;
     private static final int CONTROL_FLAG_MASK = 0x20;
+    private static final byte DELETE_HORIZON_FLAG_MASK = 0x40;
     private static final byte TIMESTAMP_TYPE_MASK = 0x08;
 
     private static final int MAX_SKIP_BUFFER_SIZE = 2048;
@@ -155,13 +160,27 @@ public void ensureValid() {
     }
 
     /**
-     * Get the timestamp of the first record in this batch. It is always the create time of the record even if the
+     * Gets the base timestamp of the batch which is used to calculate the timestamp deltas.
+     * 
+     * @return The base timestamp or
+     *         {@link RecordBatch#NO_TIMESTAMP} if the batch is empty
+     */
+    public long baseTimestamp() {
+        return buffer.getLong(FIRST_TIMESTAMP_OFFSET);
+    }
+
+    /**
+     * Get the timestamp of the first record in this batch. It is usually the create time of the record even if the
      * timestamp type of the batch is log append time.
-     *
-     * @return The first timestamp or {@link RecordBatch#NO_TIMESTAMP} if the batch is empty
+     * 
+     * @return The first timestamp if a record has been appended, unless the delete horizon has been set
+     *         {@link RecordBatch#NO_TIMESTAMP} if the batch is empty or if the delete horizon is set
      */
     public long firstTimestamp() {
-        return buffer.getLong(FIRST_TIMESTAMP_OFFSET);
+        final long baseTimestamp = baseTimestamp();
+        if (hasDeleteHorizonMs())
+            return RecordBatch.NO_TIMESTAMP;
+        return baseTimestamp;
     }
 
     @Override
@@ -245,6 +264,19 @@ public boolean isTransactional() {
         return (attributes() & TRANSACTIONAL_FLAG_MASK) > 0;
     }
 
+    @Override
+    public boolean hasDeleteHorizonMs() {
+        return (attributes() & DELETE_HORIZON_FLAG_MASK) > 0;
+    }
+
+    @Override
+    public long deleteHorizonMs() {
+        final long baseTimestamp = baseTimestamp();
+        if (hasDeleteHorizonMs())
+            return baseTimestamp;
+        return RecordBatch.NO_TIMESTAMP;
+    }
+
     @Override
     public boolean isControlBatch() {
         return (attributes() & CONTROL_FLAG_MASK) > 0;
@@ -360,7 +392,7 @@ public void setMaxTimestamp(TimestampType timestampType, long maxTimestamp) {
         if (timestampType() == timestampType && currentMaxTimestamp == maxTimestamp)
             return;
 
-        byte attributes = computeAttributes(compressionType(), timestampType, isTransactional(), isControlBatch());
+        byte attributes = computeAttributes(compressionType(), timestampType, isTransactional(), isControlBatch(), hasDeleteHorizonMs());
         buffer.putShort(ATTRIBUTES_OFFSET, attributes);
         buffer.putLong(MAX_TIMESTAMP_OFFSET, maxTimestamp);
         long crc = computeChecksum();
@@ -407,7 +439,7 @@ public int hashCode() {
     }
 
     private static byte computeAttributes(CompressionType type, TimestampType timestampType,
-                                          boolean isTransactional, boolean isControl) {
+                                          boolean isTransactional, boolean isControl, boolean isDeleteHorizonSet) {
         if (timestampType == TimestampType.NO_TIMESTAMP_TYPE)
             throw new IllegalArgumentException("Timestamp type must be provided to compute attributes for message " +
                     "format v2 and above");
@@ -419,6 +451,8 @@ private static byte computeAttributes(CompressionType type, TimestampType timest
             attributes |= COMPRESSION_CODEC_MASK & type.id;
         if (timestampType == TimestampType.LOG_APPEND_TIME)
             attributes |= TIMESTAMP_TYPE_MASK;
+        if (isDeleteHorizonSet)
+            attributes |= DELETE_HORIZON_FLAG_MASK;
         return attributes;
     }
 
@@ -436,8 +470,8 @@ public static void writeEmptyHeader(ByteBuffer buffer,
                                         boolean isControlRecord) {
         int offsetDelta = (int) (lastOffset - baseOffset);
         writeHeader(buffer, baseOffset, offsetDelta, DefaultRecordBatch.RECORD_BATCH_OVERHEAD, magic,
-                CompressionType.NONE, timestampType, RecordBatch.NO_TIMESTAMP, timestamp, producerId,
-                producerEpoch, baseSequence, isTransactional, isControlRecord, partitionLeaderEpoch, 0);
+                    CompressionType.NONE, timestampType, RecordBatch.NO_TIMESTAMP, timestamp, producerId,
+                    producerEpoch, baseSequence, isTransactional, isControlRecord, false, partitionLeaderEpoch, 0);
     }
 
     static void writeHeader(ByteBuffer buffer,
@@ -454,14 +488,15 @@ static void writeHeader(ByteBuffer buffer,
                             int sequence,
                             boolean isTransactional,
                             boolean isControlBatch,
+                            boolean isDeleteHorizonSet,
                             int partitionLeaderEpoch,
                             int numRecords) {
         if (magic < RecordBatch.CURRENT_MAGIC_VALUE)
             throw new IllegalArgumentException("Invalid magic value " + magic);
         if (firstTimestamp < 0 && firstTimestamp != NO_TIMESTAMP)
             throw new IllegalArgumentException("Invalid message timestamp " + firstTimestamp);
 
-        short attributes = computeAttributes(compressionType, timestampType, isTransactional, isControlBatch);
+        short attributes = computeAttributes(compressionType, timestampType, isTransactional, isControlBatch, isDeleteHorizonSet);
 
         int position = buffer.position();
         buffer.putLong(position + BASE_OFFSET_OFFSET, baseOffset);
@@ -699,6 +734,18 @@ public boolean isTransactional() {
             return loadBatchHeader().isTransactional();
         }
 
+        @Override
+        public boolean hasDeleteHorizonMs() {
+            return loadBatchHeader().hasDeleteHorizonMs();
+        }
+
+        @Override
+        public long deleteHorizonMs() {
+            if (hasDeleteHorizonMs())
+                return super.loadBatchHeader().deleteHorizonMs();
+            return RecordBatch.NO_TIMESTAMP;
+        }
+
         @Override
         public boolean isControlBatch() {
             return loadBatchHeader().isControlBatch();