junrao commented on code in PR #19261:
URL: https://github.com/apache/kafka/pull/19261#discussion_r2025608562
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2484,6 +2505,174 @@ private long startOffsetDuringInitialization(long
partitionDataStartOffset) thro
}
}
+ private ShareAcquiredRecords
maybeFilterAbortedTransactionalAcquiredRecords(
+ FetchPartitionData fetchPartitionData,
+ FetchIsolation isolationLevel,
+ ShareAcquiredRecords shareAcquiredRecords
+ ) {
+ if (isolationLevel != FetchIsolation.TXN_COMMITTED ||
fetchPartitionData.abortedTransactions.isEmpty() ||
fetchPartitionData.abortedTransactions.get().isEmpty())
+ return shareAcquiredRecords;
+ // When FetchIsolation.TXN_COMMITTED is used as isolation level by the
share group, we need to filter any
+ // transactions that were aborted/did not commit due to timeout.
+ List<AcquiredRecords> result =
filterAbortedTransactionalAcquiredRecords(fetchPartitionData.records.batches(),
+ shareAcquiredRecords.acquiredRecords(),
fetchPartitionData.abortedTransactions.get());
+ int acquiredCount = 0;
+ for (AcquiredRecords records : result) {
+ acquiredCount += (int) (records.lastOffset() -
records.firstOffset() + 1);
+ }
+ return new ShareAcquiredRecords(result, acquiredCount);
+ }
+
+ private List<AcquiredRecords> filterAbortedTransactionalAcquiredRecords(
+ Iterable<? extends RecordBatch> batches,
+ List<AcquiredRecords> acquiredRecords,
+ List<FetchResponseData.AbortedTransaction> abortedTransactions
+ ) {
+ lock.writeLock().lock();
+ try {
+ // The record batches that need to be archived in cachedState
because they were a part of aborted transactions.
+ List<RecordBatch> recordsToArchive =
fetchAbortedTransactionRecordBatches(batches, abortedTransactions);
+ for (RecordBatch recordBatch : recordsToArchive) {
+ // Archive the offsets/batches in the cached state.
+ NavigableMap<Long, InFlightBatch> subMap =
fetchSubMapOrException(recordBatch);
+ archiveAcquiredBatchRecords(subMap, recordBatch);
+ }
+ return filterRecordBatchesFromAcquiredRecords(acquiredRecords,
recordsToArchive);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ // Visible for testing.
+ List<AcquiredRecords> filterRecordBatchesFromAcquiredRecords(
+ List<AcquiredRecords> acquiredRecords,
+ List<RecordBatch> recordsToArchive
+ ) {
+ lock.writeLock().lock();
+ try {
+ List<AcquiredRecords> result = new ArrayList<>();
+
+ for (AcquiredRecords acquiredRecord : acquiredRecords) {
+ List<AcquiredRecords> tempAcquiredRecords = new ArrayList<>();
+ tempAcquiredRecords.add(acquiredRecord);
+ for (RecordBatch recordBatch : recordsToArchive) {
+ List<AcquiredRecords> newAcquiredRecords = new
ArrayList<>();
+ for (AcquiredRecords temp : tempAcquiredRecords) {
Review Comment:
This is probably ok, but we could have just iterated recordsToArchive and
tempAcquiredRecords once to do a merge instead of double loop.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2484,6 +2505,174 @@ private long startOffsetDuringInitialization(long
partitionDataStartOffset) thro
}
}
+ private ShareAcquiredRecords
maybeFilterAbortedTransactionalAcquiredRecords(
+ FetchPartitionData fetchPartitionData,
+ FetchIsolation isolationLevel,
+ ShareAcquiredRecords shareAcquiredRecords
+ ) {
+ if (isolationLevel != FetchIsolation.TXN_COMMITTED ||
fetchPartitionData.abortedTransactions.isEmpty() ||
fetchPartitionData.abortedTransactions.get().isEmpty())
+ return shareAcquiredRecords;
+ // When FetchIsolation.TXN_COMMITTED is used as isolation level by the
share group, we need to filter any
+ // transactions that were aborted/did not commit due to timeout.
+ List<AcquiredRecords> result =
filterAbortedTransactionalAcquiredRecords(fetchPartitionData.records.batches(),
+ shareAcquiredRecords.acquiredRecords(),
fetchPartitionData.abortedTransactions.get());
+ int acquiredCount = 0;
+ for (AcquiredRecords records : result) {
+ acquiredCount += (int) (records.lastOffset() -
records.firstOffset() + 1);
+ }
+ return new ShareAcquiredRecords(result, acquiredCount);
+ }
+
+ private List<AcquiredRecords> filterAbortedTransactionalAcquiredRecords(
+ Iterable<? extends RecordBatch> batches,
+ List<AcquiredRecords> acquiredRecords,
+ List<FetchResponseData.AbortedTransaction> abortedTransactions
+ ) {
+ lock.writeLock().lock();
Review Comment:
Do we need to lock here? The only caller is `acquire()` and already holds
the lock. Ditto for the locking in `archiveRecords()`
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2484,6 +2505,174 @@ private long startOffsetDuringInitialization(long
partitionDataStartOffset) thro
}
}
+ private ShareAcquiredRecords
maybeFilterAbortedTransactionalAcquiredRecords(
+ FetchPartitionData fetchPartitionData,
+ FetchIsolation isolationLevel,
+ ShareAcquiredRecords shareAcquiredRecords
+ ) {
+ if (isolationLevel != FetchIsolation.TXN_COMMITTED ||
fetchPartitionData.abortedTransactions.isEmpty() ||
fetchPartitionData.abortedTransactions.get().isEmpty())
+ return shareAcquiredRecords;
+ // When FetchIsolation.TXN_COMMITTED is used as isolation level by the
share group, we need to filter any
+ // transactions that were aborted/did not commit due to timeout.
+ List<AcquiredRecords> result =
filterAbortedTransactionalAcquiredRecords(fetchPartitionData.records.batches(),
+ shareAcquiredRecords.acquiredRecords(),
fetchPartitionData.abortedTransactions.get());
+ int acquiredCount = 0;
+ for (AcquiredRecords records : result) {
+ acquiredCount += (int) (records.lastOffset() -
records.firstOffset() + 1);
+ }
+ return new ShareAcquiredRecords(result, acquiredCount);
+ }
+
+ private List<AcquiredRecords> filterAbortedTransactionalAcquiredRecords(
+ Iterable<? extends RecordBatch> batches,
+ List<AcquiredRecords> acquiredRecords,
+ List<FetchResponseData.AbortedTransaction> abortedTransactions
+ ) {
+ lock.writeLock().lock();
+ try {
+ // The record batches that need to be archived in cachedState
because they were a part of aborted transactions.
+ List<RecordBatch> recordsToArchive =
fetchAbortedTransactionRecordBatches(batches, abortedTransactions);
+ for (RecordBatch recordBatch : recordsToArchive) {
+ // Archive the offsets/batches in the cached state.
+ NavigableMap<Long, InFlightBatch> subMap =
fetchSubMapOrException(recordBatch);
+ archiveAcquiredBatchRecords(subMap, recordBatch);
+ }
+ return filterRecordBatchesFromAcquiredRecords(acquiredRecords,
recordsToArchive);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ // Visible for testing.
+ List<AcquiredRecords> filterRecordBatchesFromAcquiredRecords(
+ List<AcquiredRecords> acquiredRecords,
+ List<RecordBatch> recordsToArchive
+ ) {
+ lock.writeLock().lock();
Review Comment:
Do we need to lock here since the caller already holds the lock?
--
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
