clolov commented on code in PR #14489: URL: https://github.com/apache/kafka/pull/14489#discussion_r1350305894
########## core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala: ########## @@ -392,6 +392,12 @@ class TransactionStateManager(brokerId: Int, } } + private[transaction] def hasTxnStateLoaded(partitionId: Int): Boolean = { + inReadLock(stateLock) { + !transactionMetadataCache.get(partitionId).isEmpty Review Comment: ```suggestion transactionMetadataCache.contains(partitionId) ``` ########## core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala: ########## @@ -524,23 +533,35 @@ class TransactionStateManager(brokerId: Int, } def loadTransactions(startTimeMs: java.lang.Long): Unit = { - val schedulerTimeMs = time.milliseconds() - startTimeMs - info(s"Loading transaction metadata from $topicPartition at epoch $coordinatorEpoch") - validateTransactionTopicPartitionCountIsStable() - - val loadedTransactions = loadTransactionMetadata(topicPartition, coordinatorEpoch) - val endTimeMs = time.milliseconds() - val totalLoadingTimeMs = endTimeMs - startTimeMs - partitionLoadSensor.record(totalLoadingTimeMs.toDouble, endTimeMs, false) - info(s"Finished loading ${loadedTransactions.size} transaction metadata from $topicPartition in " + - s"$totalLoadingTimeMs milliseconds, of which $schedulerTimeMs milliseconds was spent in the scheduler.") + val maybeLoadedTransactions = + if (!hadTransactionStateAlreadyLoaded) { Review Comment: ```suggestion if (!hasTransactionStateAlreadyLoaded) { ``` ########## core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala: ########## @@ -875,18 +875,35 @@ class TransactionStateManagerTest { val startOffset = 0L val records = MemoryRecords.withRecords(startOffset, CompressionType.NONE, txnRecords.toArray: _*) + val transactionsWithPendingMarkers = new ConcurrentHashMap[String, PendingCompleteTxn] + def sendMarkers(coordinatorEpoch: Int, + txnResult: TransactionResult, + txnMetadata: TransactionMetadata, + newMetadata: TxnTransitMetadata): Unit = { + val transactionalId = txnMetadata.transactionalId + val pendingCompleteTxn = PendingCompleteTxn( + transactionalId, + coordinatorEpoch, + txnMetadata, + newMetadata) + + transactionsWithPendingMarkers.put(transactionalId, pendingCompleteTxn) + } + prepareTxnLog(topicPartition, 0, records) // immigrate partition at epoch 0 - transactionManager.loadTransactionsForTxnTopicPartition(partitionId, coordinatorEpoch = 0, (_, _, _, _) => ()) + transactionManager.loadTransactionsForTxnTopicPartition(partitionId, coordinatorEpoch = 0, sendMarkers, false) Review Comment: Since this is a Scala test class can you use named arguments to improve readability similar to the argument `coordinatorEpoch = 0`? ########## metadata/src/test/java/org/apache/kafka/image/TopicsImageTest.java: ########## @@ -218,7 +218,11 @@ public void testBasicLocalChanges() { ); assertEquals( new HashSet<>(Arrays.asList(new TopicPartition("baz", 0))), - changes.leaders().keySet() + changes.electedLeaders().keySet() + ); + assertEquals( + new HashSet<>(Arrays.asList(new TopicPartition("baz", 0))), Review Comment: ```suggestion new HashSet<>(Arrays.asList(new TopicPartition("baz", 0))), ``` ########## metadata/src/test/java/org/apache/kafka/image/TopicsImageTest.java: ########## @@ -269,14 +273,49 @@ public void testDeleteAfterChanges() { LocalReplicaChanges changes = delta.localChanges(localId); assertEquals(new HashSet<>(Arrays.asList(new TopicPartition("zoo", 0))), changes.deletes()); - assertEquals(Collections.emptyMap(), changes.leaders()); + assertEquals(Collections.emptyMap(), changes.electedLeaders()); + assertEquals(Collections.emptyMap(), changes.updatedLeaders()); assertEquals(Collections.emptyMap(), changes.followers()); TopicsImage finalImage = delta.apply(); List<ApiMessageAndVersion> imageRecords = getImageRecords(image); imageRecords.addAll(topicRecords); testToImage(finalImage, Optional.of(imageRecords)); } + @Test Review Comment: ```suggestion @Test ``` ########## metadata/src/main/java/org/apache/kafka/image/LocalReplicaChanges.java: ########## @@ -26,19 +26,22 @@ public final class LocalReplicaChanges { private final Set<TopicPartition> deletes; - private final Map<TopicPartition, PartitionInfo> leaders; + private final Map<TopicPartition, PartitionInfo> electedLeaders; + private final Map<TopicPartition, PartitionInfo> updatedLeaders; Review Comment: Does it make sense to change these names to tpToPartitionEpochs and tpToLeaderEpochs? I can anticipate this naming being confusing for a person reading the code for the first time given that what classifies as each is defined in https://github.com/apache/kafka/pull/14489/files#diff-be8b1b8ad296c48bbdc3df55fdb859881f150ceadd0959ebf02fb3caac13ee5aR146-R151 ########## core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala: ########## @@ -515,7 +521,10 @@ class TransactionStateManager(brokerId: Int, * metadata cache with the transactional ids. This operation must be resilient to any partial state left off from * the previous loading / unloading operation. */ - def loadTransactionsForTxnTopicPartition(partitionId: Int, coordinatorEpoch: Int, sendTxnMarkers: SendTxnMarkersCallback): Unit = { + def loadTransactionsForTxnTopicPartition(partitionId: Int, + coordinatorEpoch: Int, + sendTxnMarkers: SendTxnMarkersCallback, + hadTransactionStateAlreadyLoaded: Boolean): Unit = { Review Comment: ```suggestion hasTransactionStateAlreadyLoaded: Boolean): Unit = { ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org