[KAFKA-8522] Streamline tombstone and transaction marker removal

checkstyle/checkstyle.xml

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

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

@@ -91,7 +91,7 @@
  * 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 time for which tombstones / transaction markers need 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:
@@ -167,21 +167,9 @@ public void ensureValid() {
      *         {@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 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() {
-        final long baseTimestamp = baseTimestamp();
         if (hasDeleteHorizonMs())
             return RecordBatch.NO_TIMESTAMP;
-        return baseTimestamp;
+        return buffer.getLong(FIRST_TIMESTAMP_OFFSET);
     }
 
     @Override
@@ -272,9 +260,8 @@ public boolean hasDeleteHorizonMs() {
 
     @Override
     public long deleteHorizonMs() {
-        final long baseTimestamp = baseTimestamp();
         if (hasDeleteHorizonMs())
-            return baseTimestamp;
+            return buffer.getLong(FIRST_TIMESTAMP_OFFSET);
         return RecordBatch.NO_TIMESTAMP;
     }
 
@@ -322,9 +309,9 @@ private CloseableIterator<Record> uncompressedIterator() {
         buffer.position(RECORDS_OFFSET);
         return new RecordIterator() {
             @Override
-            protected Record readNext(long baseOffset, long firstTimestamp, int baseSequence, Long logAppendTime) {
+            protected Record readNext(long baseOffset, long baseTimestamp, int baseSequence, Long logAppendTime) {
                 try {
-                    return DefaultRecord.readFrom(buffer, baseOffset, firstTimestamp, baseSequence, logAppendTime);
+                    return DefaultRecord.readFrom(buffer, baseOffset, baseTimestamp, baseSequence, logAppendTime);
                 } catch (BufferUnderflowException e) {
                     throw new InvalidRecordException("Incorrect declared batch size, premature EOF reached");
                 }
@@ -590,15 +577,15 @@ public static int decrementSequence(int sequence, int decrement) {
     private abstract class RecordIterator implements CloseableIterator<Record> {
         private final Long logAppendTime;
         private final long baseOffset;
-        private final long firstTimestamp;
+        private final long baseTimestamp;
         private final int baseSequence;
         private final int numRecords;
         private int readRecords = 0;
 
         RecordIterator() {
             this.logAppendTime = timestampType() == TimestampType.LOG_APPEND_TIME ? maxTimestamp() : null;
             this.baseOffset = baseOffset();
-            this.firstTimestamp = firstTimestamp();
+            this.baseTimestamp = baseTimestamp();
             this.baseSequence = baseSequence();
             int numRecords = count();
             if (numRecords < 0)
@@ -618,7 +605,7 @@ public Record next() {
                 throw new NoSuchElementException();
 
             readRecords++;
-            Record rec = readNext(baseOffset, firstTimestamp, baseSequence, logAppendTime);
+            Record rec = readNext(baseOffset, baseTimestamp, baseSequence, logAppendTime);
             if (readRecords == numRecords) {
                 // Validate that the actual size of the batch is equal to declared size
                 // by checking that after reading declared number of items, there no items left
@@ -629,7 +616,7 @@ public Record next() {
             return rec;
         }
 
-        protected abstract Record readNext(long baseOffset, long firstTimestamp, int baseSequence, Long logAppendTime);
+        protected abstract Record readNext(long baseOffset, long baseTimestamp, int baseSequence, Long logAppendTime);
 
         protected abstract boolean ensureNoneRemaining();
 
@@ -651,9 +638,9 @@ private abstract class StreamRecordIterator extends RecordIterator {
         abstract Record doReadRecord(long baseOffset, long firstTimestamp, int baseSequence, Long logAppendTime) throws IOException;
 
         @Override
-        protected Record readNext(long baseOffset, long firstTimestamp, int baseSequence, Long logAppendTime) {
+        protected Record readNext(long baseOffset, long baseTimestamp, int baseSequence, Long logAppendTime) {
             try {
-                return doReadRecord(baseOffset, firstTimestamp, baseSequence, logAppendTime);
+                return doReadRecord(baseOffset, baseTimestamp, baseSequence, logAppendTime);
             } catch (EOFException e) {
                 throw new InvalidRecordException("Incorrect declared batch size, premature EOF reached");
             } catch (IOException e) {

clients/src/test/java/org/apache/kafka/common/record/DefaultRecordBatchTest.java

@@ -67,7 +67,7 @@ public void testWriteEmptyHeader() {
                     assertEquals(isTransactional, batch.isTransactional());
                     assertEquals(timestampType, batch.timestampType());
                     assertEquals(timestamp, batch.maxTimestamp());
-                    assertEquals(RecordBatch.NO_TIMESTAMP, batch.firstTimestamp());
+                    assertEquals(RecordBatch.NO_TIMESTAMP, batch.baseTimestamp());
                     assertEquals(isControlBatch, batch.isControlBatch());
                 }
             }

core/src/test/scala/unit/kafka/log/LogValidatorTest.scala

@@ -1521,7 +1521,7 @@ class LogValidatorTest {
   def maybeCheckBaseTimestamp(expected: Long, batch: RecordBatch): Unit = {
     batch match {
       case b: DefaultRecordBatch =>
-        assertEquals(expected, b.firstTimestamp, s"Unexpected base timestamp of batch $batch")
+        assertEquals(expected, b.baseTimestamp, s"Unexpected base timestamp of batch $batch")
       case _ => // no-op
     }
   }