fix(worker): update CommitResponse to use partition type from writerFactory (#2677)

* fix(worker): update CommitResponse to use partition type from writerFactory

* fix(worker): mock partitionSpec in TopicPartitionsWorkerTest for unpartitioned partitions

* fix(worker): reorganize imports in TopicPartitionsWorkerTest for clarity

Author: Gezi-lzq

core/src/main/java/kafka/automq/table/worker/TopicPartitionsWorker.java

@@ -685,7 +685,7 @@ private CompletableFuture<Void> commitResponse(RequestWrapper requestWrapper, bo
                     .collect(Collectors.toList());
                 TopicMetric topicMetric = new TopicMetric(fieldCount);
 
-                CommitResponse commitResponse = new CommitResponse(Types.StructType.of(), fastCommit ? Errors.MORE_DATA : Errors.NONE,
+                CommitResponse commitResponse = new CommitResponse(writerFactory.partitionSpec().partitionType(), fastCommit ? Errors.MORE_DATA : Errors.NONE,
                     requestWrapper.request.commitId(), topic, nextOffsets, dataFiles, deleteFiles, topicMetric, partitionMetrics);
                 return channel.asyncSend(topic, new Event(System.currentTimeMillis(), EventType.COMMIT_RESPONSE, commitResponse))
                     .thenAccept(rst -> LOGGER.info("[COMMIT_RESPONSE],{}", commitResponse));

core/src/test/java/kafka/automq/table/worker/TopicPartitionsWorkerTest.java

@@ -36,6 +36,7 @@
 import com.automq.stream.utils.Threads;
 import com.automq.stream.utils.threads.EventLoop;
 
+import org.apache.iceberg.PartitionSpec;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
@@ -102,6 +103,7 @@ public void setup() {
         when(channel.asyncSend(eq(TOPIC), any())).thenReturn(CompletableFuture.completedFuture(null));
 
         writerFactory = mock(WriterFactory.class);
+        when(writerFactory.partitionSpec()).thenReturn(PartitionSpec.unpartitioned());
         writers = new ArrayList<>();
         when(writerFactory.newWriter()).thenAnswer(invocation -> {
             MemoryWriter writer = new MemoryWriter(config);