all test pass

Author: TaiJuWu

storage/src/test/java/org/apache/kafka/storage/internals/log/LogTestUtils.java

@@ -18,9 +18,12 @@
 
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.record.ControlRecordType;
+import org.apache.kafka.common.record.EndTransactionMarker;
 import org.apache.kafka.common.record.FileRecords;
+import org.apache.kafka.common.record.MemoryRecords;
 import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.coordinator.transaction.TransactionLogConfig;
+import org.apache.kafka.server.common.RequestLocal;
 import org.apache.kafka.server.config.ServerLogConfigs;
 import org.apache.kafka.server.util.Scheduler;
 import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
@@ -29,7 +32,6 @@
 import java.io.IOException;
 import java.util.Optional;
 import java.util.Properties;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
 public class LogTestUtils {
@@ -44,27 +46,27 @@ public static LogSegment createSegment(long offset, File logDir, int indexInterv
         return new LogSegment(ms, idx, timeIdx, txnIndex, offset, indexIntervalBytes, 0, time);
     }
 
-    public static UnifiedLog createLog(File dir,
-                                       LogConfig config,
-                                       BrokerTopicStats brokerTopicStats,
-                                       Scheduler scheduler,
-                                       Time time) throws IOException {
-        return createLog(
-                dir,
-                config,
-                brokerTopicStats,
-                scheduler,
-                time,
-                0L, // logStartOffset
-                0L, // recoveryPoint
-                5 * 60 * 1000, // maxTransactionTimeoutMs
-                new ProducerStateManagerConfig(TransactionLogConfig.PRODUCER_ID_EXPIRATION_MS_DEFAULT, false),
-                TransactionLogConfig.PRODUCER_ID_EXPIRATION_CHECK_INTERVAL_MS_DEFAULT,                true, // lastShutdownClean
-                Optional.empty(), // topicId
-                new ConcurrentHashMap<>(), // numRemainingSegments
-                false, // remoteStorageSystemEnable
-                LogOffsetsListener.NO_OP_OFFSETS_LISTENER
-        );
+    public static LogAppendInfo appendEndTxnMarkerAsLeader(UnifiedLog log,
+                                                           long producerId,
+                                                           short producerEpoch,
+                                                           ControlRecordType controlType,
+                                                           long timestamp,
+                                                           int coordinatorEpoch,
+                                                           int leaderEpoch) {
+        MemoryRecords records = endTxnRecords(controlType, producerId, producerEpoch, 0L, coordinatorEpoch, leaderEpoch, timestamp);
+
+        return log.appendAsLeader(records, leaderEpoch, AppendOrigin.COORDINATOR, RequestLocal.noCaching(), VerificationGuard.SENTINEL);
+    }
+
+    public static MemoryRecords endTxnRecords(ControlRecordType controlRecordType,
+                                              long producerId,
+                                              short epoch,
+                                              long offset,
+                                              int coordinatorEpoch,
+                                              int partitionLeaderEpoch,
+                                              long timestamp) {
+        EndTransactionMarker marker = new EndTransactionMarker(controlRecordType, coordinatorEpoch);
+        return MemoryRecords.withEndTransactionMarker(offset, timestamp, partitionLeaderEpoch, producerId, epoch, marker);
     }
 
     @SuppressWarnings("ParameterNumber")
@@ -195,7 +197,6 @@ public LogConfigBuilder withRemoteLogDeleteOnDisable(boolean remoteLogDeleteOnDi
             return this;
         }
 
-        // 3. 建立一個 build() 方法，它使用 builder 中設定的值來建立最終的 LogConfig 物件
         public LogConfig build() {
             Properties logProps = new Properties();
             logProps.put(TopicConfig.SEGMENT_MS_CONFIG, String.valueOf(segmentMs));

storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java

@@ -57,7 +57,7 @@ public class UnifiedLogTest {
     private final BrokerTopicStats brokerTopicStats = new BrokerTopicStats(false);
     private final MockTime mockTime = new MockTime();
     private final int maxTransactionTimeoutMs = 60 * 60 * 1000;
-    private final ProducerStateManagerConfig producerStateManagerConfig = new ProducerStateManagerConfig(maxTransactionTimeoutMs, true);
+    private final ProducerStateManagerConfig producerStateManagerConfig = new ProducerStateManagerConfig(maxTransactionTimeoutMs, false);
     private final List<UnifiedLog> logsToClose = new ArrayList<>();
 
     private UnifiedLog log;
@@ -564,7 +564,8 @@ public void testFirstUnstableOffsetNoTransactionalData() throws IOException {
     @Test
     public void testFirstUnstableOffsetWithTransactionalData() throws IOException {
         LogConfig logConfig = new LogTestUtils.LogConfigBuilder()
-                .withSegmentBytes(1024 * 1024 * 5).build();
+                .withSegmentBytes(1024 * 1024 * 5)
+                .build();
         log = createLog(logDir, logConfig);
 
         long pid = 137L;
@@ -588,7 +589,8 @@ public void testFirstUnstableOffsetWithTransactionalData() throws IOException {
         assertEquals(Optional.of(firstAppendInfo.firstOffset()), log.firstUnstableOffset());
 
         // now transaction is committed
-        LogAppendInfo commitAppendInfo = appendEndTxnMarkerAsLeader(log, pid, epoch, ControlRecordType.COMMIT, mockTime.milliseconds());
+        LogAppendInfo commitAppendInfo = LogTestUtils.appendEndTxnMarkerAsLeader(log, pid, epoch,
+                ControlRecordType.COMMIT, mockTime.milliseconds(), 0, 0);
 
         // first unstable offset is not updated until the high watermark is advanced
         assertEquals(Optional.of(firstAppendInfo.firstOffset()), log.firstUnstableOffset());

test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestUtils.java

@@ -24,15 +24,8 @@
 import org.apache.kafka.common.record.RecordBatch;
 import org.apache.kafka.common.record.SimpleRecord;
 import org.apache.kafka.common.record.TimestampType;
-import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.utils.Exit;
 import org.apache.kafka.common.utils.Utils;
-import org.apache.kafka.network.SocketServerConfigs;
-import org.apache.kafka.raft.QuorumConfig;
-import org.apache.kafka.server.config.KRaftConfigs;
-import org.apache.kafka.server.config.ReplicationConfigs;
-import org.apache.kafka.server.config.ServerConfigs;
-import org.apache.kafka.server.config.ServerLogConfigs;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -41,17 +34,9 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.file.Files;
-import java.util.AbstractMap;
-import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
 import java.util.Random;
-import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
 
 import static java.lang.String.format;