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
