Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -115,28 +115,25 @@ object TransactionLog {
val version = buffer.getShort
if (version >= TransactionLogValue.LOWEST_SUPPORTED_VERSION && version <= TransactionLogValue.HIGHEST_SUPPORTED_VERSION) {
val value = new TransactionLogValue(new ByteBufferAccessor(buffer), version)
val transactionMetadata = new TransactionMetadata(
val state = TransactionState.fromId(value.transactionStatus)
val tps: util.Set[TopicPartition] = new util.HashSet[TopicPartition]()
if (!state.equals(TransactionState.EMPTY))
value.transactionPartitions.forEach(partitionsSchema => {
partitionsSchema.partitionIds.forEach(partitionId => tps.add(new TopicPartition(partitionsSchema.topic, partitionId.intValue())))
})
Some(new TransactionMetadata(
transactionalId,
value.producerId,
value.previousProducerId,
value.nextProducerId,
value.producerEpoch,
RecordBatch.NO_PRODUCER_EPOCH,
value.transactionTimeoutMs,
TransactionState.fromId(value.transactionStatus),
util.Set.of(),
state,
tps,
value.transactionStartTimestampMs,
value.transactionLastUpdateTimestampMs,
TransactionVersion.fromFeatureLevel(value.clientTransactionVersion))

if (!transactionMetadata.state.equals(TransactionState.EMPTY))
value.transactionPartitions.forEach(partitionsSchema => {
transactionMetadata.addPartitions(partitionsSchema.partitionIds

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If removing addPartitions is too costly right now, could we avoid exposing it publicly and mark it VisibleForTesting instead?

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry for the late reply. I add // VisibleForTesting for functions which are only used in testing code. Thanks.

.stream
.map(partitionId => new TopicPartition(partitionsSchema.topic, partitionId.intValue()))
.toList)
})
Some(transactionMetadata)
TransactionVersion.fromFeatureLevel(value.clientTransactionVersion)))
} else throw new IllegalStateException(s"Unknown version $version from the transaction log message value")
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,7 @@ public <T> T inLock(Supplier<T> function) {
}
}

// VisibleForTesting
public void addPartitions(Collection<TopicPartition> partitions) {
topicPartitions.addAll(partitions);
}
Expand Down Expand Up @@ -500,13 +501,15 @@ public String transactionalId() {
return transactionalId;
}

// VisibleForTesting
public void setProducerId(long producerId) {
this.producerId = producerId;
}
public long producerId() {
return producerId;
}

// VisibleForTesting
public void setPrevProducerId(long prevProducerId) {
this.prevProducerId = prevProducerId;
}
Expand Down Expand Up @@ -534,6 +537,7 @@ public int txnTimeoutMs() {
return txnTimeoutMs;
}

// VisibleForTesting
public void state(TransactionState state) {
this.state = state;
}
Expand All @@ -550,6 +554,7 @@ public long txnStartTimestamp() {
return txnStartTimestamp;
}

// VisibleForTesting
public void txnLastUpdateTimestamp(long txnLastUpdateTimestamp) {
this.txnLastUpdateTimestamp = txnLastUpdateTimestamp;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
import java.util.HashSet;

/**
* Immutable object representing the target transition of the transaction metadata
* Represent the target transition of the transaction metadata. The topicPartitions field is mutable.
*/
public record TxnTransitMetadata(
long producerId,
Expand Down