This is an automated email from the ASF dual-hosted git repository.
tison pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 713b0b170bf [FLINK-28085][Connector/Pulsar] Close the pending Pulsar
transactions before closing pipeline (#21075)
713b0b170bf is described below
commit 713b0b170bf3d8d13b1663e73c1e9d6f100da731
Author: Yufan Sheng <[email protected]>
AuthorDate: Tue Oct 18 20:23:24 2022 +0800
[FLINK-28085][Connector/Pulsar] Close the pending Pulsar transactions
before closing pipeline (#21075)
---
.../common/utils/PulsarTransactionUtils.java | 2 ++
.../reader/fetcher/PulsarFetcherManagerBase.java | 12 ++++++++++
.../reader/source/PulsarOrderedSourceReader.java | 3 +++
.../reader/source/PulsarSourceReaderBase.java | 7 ++++++
.../reader/source/PulsarUnorderedSourceReader.java | 27 ++++++++++++++++++++++
.../split/PulsarUnorderedPartitionSplitReader.java | 14 +++++++----
6 files changed, 61 insertions(+), 4 deletions(-)
diff --git
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarTransactionUtils.java
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarTransactionUtils.java
index a48b4d4a7bf..c8ffe6387fd 100644
---
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarTransactionUtils.java
+++
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarTransactionUtils.java
@@ -60,6 +60,8 @@ public final class PulsarTransactionUtils {
/**
* This is a bug in original {@link
TransactionCoordinatorClientException#unwrap(Throwable)}
* method. Pulsar wraps the {@link ExecutionException} which hides the
real execution exception.
+ *
+ * <p>This bug should be fixed after the 2.10.0 release. We just keep this
for safety.
*/
public static TransactionCoordinatorClientException unwrap(
TransactionCoordinatorClientException e) {
diff --git
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java
index 6abb8bd70fc..1622eee33b8 100644
---
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java
+++
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java
@@ -86,6 +86,18 @@ public abstract class PulsarFetcherManagerBase<T>
}
}
+ /** Close the finished split related fetcher. */
+ public void closeFetcher(String splitId) {
+ Integer fetchId = splitFetcherMapping.remove(splitId);
+ if (fetchId != null) {
+ fetcherStatus.remove(fetchId);
+ SplitFetcher<PulsarMessage<T>, PulsarPartitionSplit> fetcher =
fetchers.remove(fetchId);
+ if (fetcher != null) {
+ fetcher.shutdown();
+ }
+ }
+ }
+
protected SplitFetcher<PulsarMessage<T>, PulsarPartitionSplit>
getOrCreateFetcher(
String splitId) {
SplitFetcher<PulsarMessage<T>, PulsarPartitionSplit> fetcher;
diff --git
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReader.java
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReader.java
index a06d8a47a25..6ff04666cca 100644
---
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReader.java
+++
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReader.java
@@ -109,6 +109,9 @@ public class PulsarOrderedSourceReader<OUT> extends
PulsarSourceReaderBase<OUT>
@Override
protected void onSplitFinished(Map<String, PulsarPartitionSplitState>
finishedSplitIds) {
+ // Close all the finished splits.
+ closeFinishedSplits(finishedSplitIds.keySet());
+
// We don't require new splits, all the splits are pre-assigned by
source enumerator.
if (LOG.isDebugEnabled()) {
LOG.debug("onSplitFinished event: {}", finishedSplitIds);
diff --git
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java
index c910e9461c9..eee6950af7c 100644
---
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java
+++
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java
@@ -33,6 +33,7 @@ import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.PulsarClient;
import java.util.Collection;
+import java.util.Set;
/**
* The common pulsar source reader for both ordered & unordered message
consuming.
@@ -92,4 +93,10 @@ abstract class PulsarSourceReaderBase<OUT>
pulsarClient.shutdown();
pulsarAdmin.close();
}
+
+ protected void closeFinishedSplits(Set<String> finishedSplitIds) {
+ for (String splitId : finishedSplitIds) {
+ ((PulsarFetcherManagerBase<OUT>)
splitFetcherManager).closeFetcher(splitId);
+ }
+ }
}
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 1767210b9ff..3bb4bc61b0a 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
@@ -118,6 +118,9 @@ public class PulsarUnorderedSourceReader<OUT> extends
PulsarSourceReaderBase<OUT
@Override
protected void onSplitFinished(Map<String, PulsarPartitionSplitState>
finishedSplitIds) {
+ // Close all the finished splits.
+ closeFinishedSplits(finishedSplitIds.keySet());
+
// We don't require new splits, all the splits are pre-assigned by
source enumerator.
if (LOG.isDebugEnabled()) {
LOG.debug("onSplitFinished event: {}", finishedSplitIds);
@@ -151,6 +154,11 @@ public class PulsarUnorderedSourceReader<OUT> extends
PulsarSourceReaderBase<OUT
txnIDs.add(uncommittedTransactionId);
}
}
+
+ // Add finished splits' transactions.
+ txnIDs.addAll(transactionsOfFinishedSplits);
+ // Purge the transactions.
+ transactionsOfFinishedSplits.clear();
}
return splits;
@@ -177,4 +185,23 @@ public class PulsarUnorderedSourceReader<OUT> extends
PulsarSourceReaderBase<OUT
}
}
}
+
+ @Override
+ public void close() throws Exception {
+ // Abort all the pending transactions.
+ if (coordinatorClient != null) {
+ for (List<TxnID> transactions : transactionsToCommit.values()) {
+ for (TxnID transaction : transactions) {
+ try {
+ coordinatorClient.abort(transaction);
+ } catch (Exception e) {
+ LOG.warn("Error in aborting transaction {}",
transaction, e);
+ }
+ }
+ }
+ }
+
+ // Close the pulsar client finally.
+ super.close();
+ }
}
diff --git
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java
index b5809ef9b7b..02fd8d9484e 100644
---
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java
+++
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java
@@ -33,6 +33,7 @@ import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.transaction.Transaction;
import org.apache.pulsar.client.api.transaction.TransactionCoordinatorClient;
import
org.apache.pulsar.client.api.transaction.TransactionCoordinatorClientException;
+import
org.apache.pulsar.client.api.transaction.TransactionCoordinatorClientException.TransactionNotFoundException;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,6 +47,7 @@ import java.util.concurrent.TimeUnit;
import static
org.apache.flink.connector.pulsar.common.utils.PulsarExceptionUtils.sneakyClient;
import static
org.apache.flink.connector.pulsar.common.utils.PulsarTransactionUtils.createTransaction;
+import static
org.apache.flink.connector.pulsar.common.utils.PulsarTransactionUtils.unwrap;
/**
* The split reader a given {@link PulsarPartitionSplit}, it would be closed
once the {@link
@@ -124,10 +126,14 @@ public class PulsarUnorderedPartitionSplitReader<OUT>
extends PulsarPartitionSpl
try {
coordinatorClient.abort(uncommittedTransactionId);
} catch (TransactionCoordinatorClientException e) {
- LOG.error(
- "Failed to abort the uncommitted transaction {}
when restart the reader",
- uncommittedTransactionId,
- e);
+ TransactionCoordinatorClientException exception =
unwrap(e);
+ // The aborted transaction would return a not found
exception.
+ if (!(exception instanceof TransactionNotFoundException)) {
+ LOG.error(
+ "Failed to abort the uncommitted transaction
{} when restart the reader",
+ uncommittedTransactionId,
+ e);
+ }
}
}
