You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ti...@apache.org on 2022/09/14 06:02:17 UTC
[flink] 04/06: [FLINK-27611][Connector/pulsar] Fix ConcurrentModificationException during checkpoint on Pulsar unordered reader.
This is an automated email from the ASF dual-hosted git repository.
tison pushed a commit to branch release-1.16
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4412ba7a13d82225f3082e5ee2f475f4be572d84
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Mon Sep 5 23:59:52 2022 +0800
[FLINK-27611][Connector/pulsar] Fix ConcurrentModificationException during checkpoint on Pulsar unordered reader.
---
.../reader/source/PulsarUnorderedSourceReader.java | 26 ++++++++++++----------
1 file changed, 14 insertions(+), 12 deletions(-)
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarUnorderedSourceReader.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarUnorderedSourceReader.java
index 09937d8ea05..b20b117652d 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarUnorderedSourceReader.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarUnorderedSourceReader.java
@@ -46,6 +46,8 @@ import java.util.SortedMap;
import java.util.TreeMap;
import java.util.function.Supplier;
+import static java.util.stream.Collectors.toList;
+
/**
* The source reader for pulsar subscription Shared and Key_Shared, which consumes the unordered
* messages.
@@ -159,19 +161,19 @@ public class PulsarUnorderedSourceReader<OUT> extends PulsarSourceReaderBase<OUT
LOG.debug("Committing transactions for checkpoint {}", checkpointId);
if (coordinatorClient != null) {
- for (Map.Entry<Long, List<TxnID>> entry : transactionsToCommit.entrySet()) {
- Long currentCheckpointId = entry.getKey();
- if (currentCheckpointId > checkpointId) {
- continue;
- }
-
- List<TxnID> transactions = entry.getValue();
- for (TxnID transaction : transactions) {
- coordinatorClient.commit(transaction);
- transactionsOfFinishedSplits.remove(transaction);
+ List<Long> checkpointIds =
+ transactionsToCommit.keySet().stream()
+ .filter(id -> id <= checkpointId)
+ .collect(toList());
+
+ for (Long id : checkpointIds) {
+ List<TxnID> transactions = transactionsToCommit.remove(id);
+ if (transactions != null) {
+ for (TxnID transaction : transactions) {
+ coordinatorClient.commit(transaction);
+ transactionsOfFinishedSplits.remove(transaction);
+ }
}
-
- transactionsToCommit.remove(currentCheckpointId);
}
}
}