fix tests

Author: polyzos

fluss-common/src/main/java/org/apache/fluss/record/AbstractRowMemoryLogRecordsBuilder.java

@@ -26,7 +26,6 @@
 
 import java.io.IOException;
 
-import static org.apache.fluss.record.DefaultValueRecordBatch.RECORD_BATCH_HEADER_SIZE;
 import static org.apache.fluss.record.LogRecordBatch.CURRENT_LOG_MAGIC_VALUE;
 import static org.apache.fluss.record.LogRecordBatchFormat.BASE_OFFSET_LENGTH;
 import static org.apache.fluss.record.LogRecordBatchFormat.LENGTH_LENGTH;
@@ -36,6 +35,7 @@
 import static org.apache.fluss.record.LogRecordBatchFormat.NO_WRITER_ID;
 import static org.apache.fluss.record.LogRecordBatchFormat.crcOffset;
 import static org.apache.fluss.record.LogRecordBatchFormat.lastOffsetDeltaOffset;
+import static org.apache.fluss.record.LogRecordBatchFormat.recordBatchHeaderSize;
 import static org.apache.fluss.record.LogRecordBatchFormat.schemaIdOffset;
 import static org.apache.fluss.utils.Preconditions.checkArgument;
 
@@ -82,8 +82,9 @@ protected AbstractRowMemoryLogRecordsBuilder(
         this.isClosed = false;
 
         // Skip header initially; will be written in build()
-        this.pagedOutputView.setPosition(RECORD_BATCH_HEADER_SIZE);
-        this.sizeInBytes = RECORD_BATCH_HEADER_SIZE;
+        int headerSize = recordBatchHeaderSize(magic);
+        this.pagedOutputView.setPosition(headerSize);
+        this.sizeInBytes = headerSize;
     }
 
     protected AbstractRowMemoryLogRecordsBuilder(

fluss-common/src/main/java/org/apache/fluss/record/DefaultLogRecordBatch.java

@@ -34,7 +34,6 @@
 import java.nio.ByteBuffer;
 import java.util.NoSuchElementException;
 
-import static org.apache.fluss.record.DefaultValueRecordBatch.RECORD_BATCH_HEADER_SIZE;
 import static org.apache.fluss.record.LogRecordBatchFormat.BASE_OFFSET_OFFSET;
 import static org.apache.fluss.record.LogRecordBatchFormat.COMMIT_TIMESTAMP_OFFSET;
 import static org.apache.fluss.record.LogRecordBatchFormat.LENGTH_OFFSET;
@@ -285,7 +284,7 @@ private CloseableIterator<LogRecord> compactedRowRecordIterator(
             RowType rowType, long timestamp) {
         DataType[] fieldTypes = rowType.getChildren().toArray(new DataType[0]);
         return new LogRecordIterator() {
-            int position = DefaultLogRecordBatch.this.position + RECORD_BATCH_HEADER_SIZE;
+            int position = DefaultLogRecordBatch.this.position + recordBatchHeaderSize(magic);
             int rowId = 0;
 
             @Override

fluss-common/src/test/java/org/apache/fluss/record/MemoryLogRecordsCompactedBuilderTest.java

@@ -115,7 +115,10 @@ void testNoRecordAppendAndBaseOffset() throws Exception {
         // base offset 0
         MemoryLogRecordsCompactedBuilder builder = createBuilder(0, 1, 1024);
         MemoryLogRecords records = MemoryLogRecords.pointToBytesView(builder.build());
-        assertThat(records.sizeInBytes()).isEqualTo(48); // only batch header
+        assertThat(records.sizeInBytes())
+                .isEqualTo(
+                        LogRecordBatchFormat.recordBatchHeaderSize(
+                                DEFAULT_MAGIC)); // only batch header
         LogRecordBatch batch = records.batches().iterator().next();
         batch.ensureValid();
         assertThat(batch.getRecordCount()).isEqualTo(0);
@@ -132,7 +135,8 @@ void testNoRecordAppendAndBaseOffset() throws Exception {
         // base offset 100
         builder = createBuilder(100, 1, 1024);
         records = MemoryLogRecords.pointToBytesView(builder.build());
-        assertThat(records.sizeInBytes()).isEqualTo(48);
+        assertThat(records.sizeInBytes())
+                .isEqualTo(LogRecordBatchFormat.recordBatchHeaderSize(DEFAULT_MAGIC));
         batch = records.batches().iterator().next();
         batch.ensureValid();
         assertThat(batch.getRecordCount()).isEqualTo(0);