fix rebase compilation issues

Author: polyzos

fluss-client/src/main/java/org/apache/fluss/client/write/CompactedLogWriteBatch.java

@@ -118,6 +118,11 @@ public int getSizeInBytes() {
                 "Failed to build compacted log record batch.");
     }
 
+    @Override
+    public boolean isLogBatch() {
+        return true;
+    }
+
     @Override
     protected CompactedRow requireAndCastRow(InternalRow row) {
         checkArgument(

fluss-client/src/main/java/org/apache/fluss/client/write/IndexedLogWriteBatch.java

@@ -59,35 +59,6 @@ public IndexedLogWriteBatch(
                             MemoryLogRecordsIndexedBuilder.builder(
                                     schemaId, writeLimit, outputView, true);
 
-                    @Override
-                    public boolean isLogBatch() {
-                        return true;
-                    }
-
-                    @Override
-                    public boolean tryAppend(WriteRecord writeRecord, WriteCallback callback)
-                            throws Exception {
-                        checkNotNull(callback, "write callback must be not null");
-                        checkNotNull(writeRecord.getRow(), "row must not be null for log record");
-                        checkArgument(
-                                writeRecord.getKey() == null, "key must be null for log record");
-                        checkArgument(
-                                writeRecord.getTargetColumns() == null,
-                                "target columns must be null for log record");
-                        checkArgument(
-                                writeRecord.getRow() instanceof IndexedRow,
-                                "row must not be IndexRow for indexed log table");
-                        IndexedRow row = (IndexedRow) writeRecord.getRow();
-                        if (!recordsBuilder.hasRoomFor(row) || isClosed()) {
-                            return false;
-                        } else {
-                            recordsBuilder.append(ChangeType.APPEND_ONLY, row);
-                            recordCount++;
-                            callbacks.add(callback);
-                            return true;
-                        }
-                    }
-
                     @Override
                     public boolean hasRoomFor(IndexedRow row) {
                         return delegate.hasRoomFor(row);
@@ -146,6 +117,11 @@ public int getSizeInBytes() {
                 "Failed to build indexed log record batch.");
     }
 
+    @Override
+    public boolean isLogBatch() {
+        return true;
+    }
+
     @Override
     protected IndexedRow requireAndCastRow(org.apache.fluss.row.InternalRow row) {
         checkArgument(row instanceof IndexedRow, "row must be IndexRow for indexed log table");

fluss-client/src/main/java/org/apache/fluss/client/write/WriteRecord.java

@@ -129,10 +129,14 @@ public static WriteRecord forArrowAppend(
 
     /** Creates a write record for append operation for Compacted format. */
     public static WriteRecord forCompactedAppend(
-            PhysicalTablePath tablePath, CompactedRow row, @Nullable byte[] bucketKey) {
+            TableInfo tableInfo,
+            PhysicalTablePath tablePath,
+            CompactedRow row,
+            @Nullable byte[] bucketKey) {
         checkNotNull(row);
         int estimatedSizeInBytes = CompactedLogRecord.sizeOf(row) + RECORD_BATCH_HEADER_SIZE;
         return new WriteRecord(
+                tableInfo,
                 tablePath,
                 null,
                 bucketKey,

fluss-client/src/test/java/org/apache/fluss/client/write/CompactedLogWriteBatchTest.java

@@ -198,7 +198,8 @@ void testBatchAborted() throws Exception {
     }
 
     private WriteRecord createWriteRecord() {
-        return WriteRecord.forCompactedAppend(DATA1_PHYSICAL_TABLE_PATH, row, null);
+        return WriteRecord.forCompactedAppend(
+                DATA1_TABLE_INFO, DATA1_PHYSICAL_TABLE_PATH, row, null);
     }
 
     private CompactedLogWriteBatch createLogWriteBatch(TableBucket tb, long baseLogOffset)