This is an automated email from the ASF dual-hosted git repository.
ferenc-csaky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
The following commit(s) were added to refs/heads/main by this push:
new e26b600 [FLINK-41009] Use a single transaction per checkpoint
e26b600 is described below
commit e26b60037d59474fa7c15b7783953a2a01654d54
Author: fengyubiao <[email protected]>
AuthorDate: Sun Oct 25 14:04:36 2026 +0800
[FLINK-41009] Use a single transaction per checkpoint
---
.../pulsar/sink/committer/MessageIdPojo.java | 119 ++++++++++++++++
.../pulsar/sink/committer/PulsarCommittable.java | 29 ++--
.../committer/PulsarCommittableSerializer.java | 73 ++++++++--
.../pulsar/sink/committer/PulsarCommitter.java | 7 +-
.../connector/pulsar/sink/writer/PulsarWriter.java | 67 +++++++--
.../pulsar/sink/writer/topic/ProducerRegister.java | 63 ++++-----
.../committer/PulsarCommittableSerializerTest.java | 152 ++++++++++++++++++++-
.../pulsar/sink/writer/PulsarWriterTest.java | 1 +
.../sink/writer/topic/ProducerRegisterTest.java | 40 ++++--
9 files changed, 461 insertions(+), 90 deletions(-)
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/MessageIdPojo.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/MessageIdPojo.java
new file mode 100644
index 0000000..9984904
--- /dev/null
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/MessageIdPojo.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.sink.committer;
+
+import org.apache.flink.annotation.Internal;
+
+import java.util.Objects;
+
+/** A POJO representation of Pulsar message ID used for serialization. */
+@Internal
+public class MessageIdPojo {
+
+ private int batchIndex;
+ private int batchSize;
+ private long ledgerId;
+ private long entryId;
+ private int partitionIndex;
+
+ public MessageIdPojo() {}
+
+ public MessageIdPojo(
+ long ledgerId, long entryId, int batchSize, int batchIndex, int
partitionIndex) {
+ this.batchIndex = batchIndex;
+ this.batchSize = batchSize;
+ this.ledgerId = ledgerId;
+ this.entryId = entryId;
+ this.partitionIndex = partitionIndex;
+ }
+
+ public int getBatchIndex() {
+ return batchIndex;
+ }
+
+ public void setBatchIndex(int batchIndex) {
+ this.batchIndex = batchIndex;
+ }
+
+ public int getBatchSize() {
+ return batchSize;
+ }
+
+ public void setBatchSize(int batchSize) {
+ this.batchSize = batchSize;
+ }
+
+ public long getLedgerId() {
+ return ledgerId;
+ }
+
+ public void setLedgerId(long ledgerId) {
+ this.ledgerId = ledgerId;
+ }
+
+ public long getEntryId() {
+ return entryId;
+ }
+
+ public void setEntryId(long entryId) {
+ this.entryId = entryId;
+ }
+
+ public int getPartitionIndex() {
+ return partitionIndex;
+ }
+
+ public void setPartitionIndex(int partitionIndex) {
+ this.partitionIndex = partitionIndex;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof MessageIdPojo)) {
+ return false;
+ }
+ MessageIdPojo that = (MessageIdPojo) o;
+ return batchIndex == that.batchIndex
+ && batchSize == that.batchSize
+ && ledgerId == that.ledgerId
+ && entryId == that.entryId
+ && partitionIndex == that.partitionIndex;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(batchIndex, batchSize, ledgerId, entryId,
partitionIndex);
+ }
+
+ @Override
+ public String toString() {
+ return "MessageIdPojo{"
+ + "batchIndex="
+ + batchIndex
+ + ", batchSize="
+ + batchSize
+ + ", ledgerId="
+ + ledgerId
+ + ", entryId="
+ + entryId
+ + ", partitionIndex="
+ + partitionIndex
+ + '}';
+ }
+}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittable.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittable.java
index cca8e80..7f33e75 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittable.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittable.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.pulsar.client.api.transaction.TxnID;
+import java.util.Map;
import java.util.Objects;
/** The writer state for Pulsar connector. We would used in Pulsar committer.
*/
@@ -31,41 +32,43 @@ public class PulsarCommittable {
/** The transaction id. */
private final TxnID txnID;
- /** The topic name with partition information. */
- private final String topic;
+ private final Map<String, MessageIdPojo> latestPublishedMessages;
- public PulsarCommittable(TxnID txnID, String topic) {
+ public PulsarCommittable(TxnID txnID, Map<String, MessageIdPojo>
latestPublishedMessages) {
this.txnID = txnID;
- this.topic = topic;
+ this.latestPublishedMessages = latestPublishedMessages;
}
public TxnID getTxnID() {
return txnID;
}
- public String getTopic() {
- return topic;
+ public Map<String, MessageIdPojo> getLatestPublishedMessages() {
+ return latestPublishedMessages;
}
@Override
public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
+ if (!(o instanceof PulsarCommittable)) {
return false;
}
PulsarCommittable that = (PulsarCommittable) o;
- return Objects.equals(txnID, that.txnID) && Objects.equals(topic,
that.topic);
+ return Objects.equals(txnID, that.txnID)
+ && Objects.equals(latestPublishedMessages,
that.latestPublishedMessages);
}
@Override
public int hashCode() {
- return Objects.hash(txnID, topic);
+ return Objects.hash(txnID, latestPublishedMessages);
}
@Override
public String toString() {
- return "PulsarCommittable{" + "txnID=" + txnID + ", topic='" + topic +
'\'' + '}';
+ return "PulsarCommittable{"
+ + "txnID="
+ + txnID
+ + ", latestPublishedMessages="
+ + latestPublishedMessages
+ + '}';
}
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializer.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializer.java
index 324a7c6..ed183c6 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializer.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializer.java
@@ -27,11 +27,20 @@ import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
-/** A serializer used to serialize {@link PulsarCommittable}. */
+/**
+ * A serializer used to serialize {@link PulsarCommittable}.
+ *
+ * <p>Each version only appends fields to the previous one, so connector
versions that only know V1
+ * can still read V2 state.
+ */
public class PulsarCommittableSerializer implements
SimpleVersionedSerializer<PulsarCommittable> {
- private static final int CURRENT_VERSION = 1;
+ private static final int CURRENT_VERSION = 2;
+ private static final String V2_LEGACY_TOPIC =
"unknown_due_to_v2_serialization";
@Override
public int getVersion() {
@@ -40,26 +49,70 @@ public class PulsarCommittableSerializer implements
SimpleVersionedSerializer<Pu
@Override
public byte[] serialize(PulsarCommittable obj) throws IOException {
- try (final ByteArrayOutputStream baos = new ByteArrayOutputStream();
- final DataOutputStream out = new DataOutputStream(baos)) {
+ try (ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream output = new DataOutputStream(bytes)) {
TxnID txnID = obj.getTxnID();
- out.writeLong(txnID.getMostSigBits());
- out.writeLong(txnID.getLeastSigBits());
- out.writeUTF(obj.getTopic());
- out.flush();
- return baos.toByteArray();
+ output.writeLong(txnID.getMostSigBits());
+ output.writeLong(txnID.getLeastSigBits());
+ // V1 topic field. A transaction can now span topics, so use a
descriptive placeholder.
+ output.writeUTF(V2_LEGACY_TOPIC);
+
+ Map<String, MessageIdPojo> messages =
obj.getLatestPublishedMessages();
+ output.writeInt(messages.size());
+ for (Map.Entry<String, MessageIdPojo> entry : messages.entrySet())
{
+ MessageIdPojo messageId = entry.getValue();
+ output.writeUTF(entry.getKey());
+ output.writeLong(messageId.getLedgerId());
+ output.writeLong(messageId.getEntryId());
+ output.writeInt(messageId.getBatchSize());
+ output.writeInt(messageId.getBatchIndex());
+ output.writeInt(messageId.getPartitionIndex());
+ }
+ output.flush();
+ return bytes.toByteArray();
}
}
@Override
public PulsarCommittable deserialize(int version, byte[] serialized)
throws IOException {
+ if (version < 1 || version > CURRENT_VERSION) {
+ throw new IOException("Unsupported Pulsar committable version: " +
version);
+ }
try (final ByteArrayInputStream bais = new
ByteArrayInputStream(serialized);
final DataInputStream in = new DataInputStream(bais)) {
long mostSigBits = in.readLong();
long leastSigBits = in.readLong();
TxnID txnID = new TxnID(mostSigBits, leastSigBits);
+ // V1 topic field, unused since a transaction can span topics.
String topic = in.readUTF();
- return new PulsarCommittable(txnID, topic);
+ if (topic.isEmpty()) {
+ throw new IOException("Corrupt V1 committable: topic must not
be empty.");
+ }
+
+ Map<String, MessageIdPojo> messages = Collections.emptyMap();
+ if (version >= CURRENT_VERSION) {
+ int messageCount = in.readInt();
+ if (messageCount < 0) {
+ throw new IOException("Negative number of published
messages: " + messageCount);
+ }
+
+ messages = new HashMap<>();
+ for (int i = 0; i < messageCount; i++) {
+ String messageTopic = in.readUTF();
+ MessageIdPojo messageId =
+ new MessageIdPojo(
+ in.readLong(),
+ in.readLong(),
+ in.readInt(),
+ in.readInt(),
+ in.readInt());
+ messages.put(messageTopic, messageId);
+ }
+ }
+ if (in.available() != 0) {
+ throw new IOException("Trailing bytes in Pulsar committable
state");
+ }
+ return new PulsarCommittable(txnID, messages);
}
}
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommitter.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommitter.java
index 4b7506c..d3e6b63 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommitter.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommitter.java
@@ -75,9 +75,7 @@ public class PulsarCommitter implements
Committer<PulsarCommittable>, Closeable
for (CommitRequest<PulsarCommittable> request : requests) {
PulsarCommittable committable = request.getCommittable();
TxnID txnID = committable.getTxnID();
- String topic = committable.getTopic();
-
- LOG.debug("Start committing the Pulsar transaction {} for topic
{}", txnID, topic);
+ LOG.debug("Start committing the Pulsar transaction {}", txnID);
try {
client.commit(txnID);
} catch (CoordinatorNotFoundException e) {
@@ -122,9 +120,8 @@ public class PulsarCommitter implements
Committer<PulsarCommittable>, Closeable
request.signalFailedWithKnownReason(e);
} catch (TransactionCoordinatorClientException e) {
LOG.error(
- "Encountered retriable exception while committing
transaction {} for topic {}.",
+ "Encountered retriable exception while committing
transaction {}.",
committable,
- topic,
e);
int maxRecommitTimes = sinkConfiguration.getMaxRecommitTimes();
if (request.getNumberOfRetries() < maxRecommitTimes) {
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
index 021a8f5..00e4245 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
@@ -26,6 +26,7 @@ import
org.apache.flink.api.connector.sink2.CommittingSinkWriter;
import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.pulsar.common.crypto.PulsarCrypto;
+import org.apache.flink.connector.pulsar.sink.committer.MessageIdPojo;
import org.apache.flink.connector.pulsar.sink.committer.PulsarCommittable;
import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
import org.apache.flink.connector.pulsar.sink.writer.context.PulsarSinkContext;
@@ -41,9 +42,11 @@ import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageIdAdv;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.TypedMessageBuilder;
+import org.apache.pulsar.client.api.transaction.TxnID;
import org.apache.pulsar.client.impl.TypedMessageBuilderImpl;
import org.apache.pulsar.shade.com.google.common.base.Strings;
import org.slf4j.Logger;
@@ -51,9 +54,12 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import static java.util.Collections.emptyList;
@@ -79,6 +85,7 @@ public class PulsarWriter<IN> implements
CommittingSinkWriter<IN, PulsarCommitta
private final ProducerRegister producerRegister;
private final MailboxExecutor mailboxExecutor;
private final AtomicLong pendingMessages;
+ private final ConcurrentHashMap<String, MessageIdAdv>
latestPublishedMessages;
/**
* Constructor creating a Pulsar writer.
@@ -136,6 +143,10 @@ public class PulsarWriter<IN> implements
CommittingSinkWriter<IN, PulsarCommitta
this.producerRegister = new ProducerRegister(sinkConfiguration,
pulsarCrypto, metricGroup);
this.mailboxExecutor = initContext.getMailboxExecutor();
this.pendingMessages = new AtomicLong(0);
+ this.latestPublishedMessages =
+ deliveryGuarantee == DeliveryGuarantee.EXACTLY_ONCE
+ ? new ConcurrentHashMap<>()
+ : null;
}
@Override
@@ -167,18 +178,38 @@ public class PulsarWriter<IN> implements
CommittingSinkWriter<IN, PulsarCommitta
CompletableFuture<MessageId> future = builder.sendAsync();
future.whenComplete(
(id, ex) -> {
- pendingMessages.decrementAndGet();
- if (ex != null) {
- mailboxExecutor.execute(
- () -> throwSendingException(topic, ex),
- "Failed to send data to Pulsar");
- } else {
- LOG.debug("Sent message to Pulsar {} with message
id {}", topic, id);
+ try {
+ if (ex != null) {
+ mailboxExecutor.execute(
+ () -> throwSendingException(topic, ex),
+ "Failed to send data to Pulsar");
+ } else {
+ recordLatestMessageId(topic, id);
+ LOG.debug(
+ "Sent message to Pulsar {} with
message id {}", topic, id);
+ }
+ } finally {
+ pendingMessages.decrementAndGet();
}
});
}
}
+ private void recordLatestMessageId(String topic, MessageId messageId) {
+ if (deliveryGuarantee != DeliveryGuarantee.EXACTLY_ONCE) {
+ return;
+ }
+
+ // Producer sends return MessageIdImpl subclasses, which implement
MessageIdAdv.
+ MessageIdAdv messageIdAdv = (MessageIdAdv) messageId;
+ latestPublishedMessages.compute(
+ topic,
+ (ignored, current) ->
+ current == null || messageIdAdv.compareTo(current) > 0
+ ? messageIdAdv
+ : current);
+ }
+
private void throwSendingException(String topic, Throwable ex) {
throw new FlinkRuntimeException("Failed to send data to Pulsar: " +
topic, ex);
}
@@ -266,7 +297,27 @@ public class PulsarWriter<IN> implements
CommittingSinkWriter<IN, PulsarCommitta
@Override
public Collection<PulsarCommittable> prepareCommit() {
if (deliveryGuarantee == DeliveryGuarantee.EXACTLY_ONCE) {
- return producerRegister.prepareCommit();
+ TxnID txnID = producerRegister.prepareCommit();
+ if (txnID == null) {
+ return Collections.emptyList();
+ }
+
+ // Flink invokes flush() before prepareCommit(). flush() waits for
all pending async
+ // sends and their callbacks, so no message ID can be added
between copying and clearing
+ // the map below.
+ Map<String, MessageIdPojo> messageIds = new HashMap<>();
+ latestPublishedMessages.forEach(
+ (topic, messageId) ->
+ messageIds.put(
+ topic,
+ new MessageIdPojo(
+ messageId.getLedgerId(),
+ messageId.getEntryId(),
+ messageId.getBatchSize(),
+ messageId.getBatchIndex(),
+ messageId.getPartitionIndex())));
+ latestPublishedMessages.clear();
+ return Collections.singletonList(new PulsarCommittable(txnID,
messageIds));
} else {
return emptyList();
}
diff --git
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegister.java
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegister.java
index b84d6d9..375f589 100644
---
a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegister.java
+++
b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegister.java
@@ -23,7 +23,6 @@ import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.pulsar.common.crypto.PulsarCrypto;
import
org.apache.flink.connector.pulsar.common.metrics.ProducerMetricsInterceptor;
import org.apache.flink.connector.pulsar.common.schema.PulsarSchemaUtils;
-import org.apache.flink.connector.pulsar.sink.committer.PulsarCommittable;
import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
@@ -56,9 +55,7 @@ import javax.annotation.Nullable;
import java.io.Closeable;
import java.io.IOException;
-import java.util.ArrayList;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
@@ -104,7 +101,7 @@ public class ProducerRegister implements Closeable {
private final SinkWriterMetricGroup metricGroup;
private final Map<String, Schema<byte[]>> schemas;
private final Map<String, Map<SchemaHash, Producer<?>>> producers;
- private final Map<String, Transaction> transactions;
+ @Nullable private Transaction transaction;
public ProducerRegister(
SinkConfiguration sinkConfiguration,
@@ -117,7 +114,6 @@ public class ProducerRegister implements Closeable {
this.metricGroup = metricGroup;
this.schemas = new HashMap<>();
this.producers = new HashMap<>();
- this.transactions = new HashMap<>();
if (sinkConfiguration.isEnableMetrics()) {
metricGroup.setCurrentSendTimeGauge(this::currentSendTimeGauge);
@@ -133,9 +129,9 @@ public class ProducerRegister implements Closeable {
/**
* Create a TypedMessageBuilder which could be sent to Pulsar directly.
First, we would create a
- * topic-related producer or use a cached one instead. Then we would try
to find a topic-related
- * transaction. We would generate a transaction instance if there is no
transaction. Finally, we
- * created the message builder and put the element into it.
+ * topic-related producer or use a cached one instead. Then we would get
the transaction for the
+ * current checkpoint, creating it if necessary. Finally, we create the
message builder and put
+ * the element into it.
*
* <p>Pulsar's Producer doesn't have {@code producer.newMessage(schema,
transaction)} method. We
* have to manually create it.
@@ -150,29 +146,23 @@ public class ProducerRegister implements Closeable {
TransactionImpl transaction = null;
if (sinkConfiguration.getDeliveryGuarantee() ==
DeliveryGuarantee.EXACTLY_ONCE) {
- transaction = (TransactionImpl) getOrCreateTransaction(topic);
+ transaction = (TransactionImpl) getOrCreateTransaction();
}
return (TypedMessageBuilder<T>)
new TypedMessageBuilderImpl<>(producer, schema, transaction);
}
- /**
- * Convert the transactions into a committable list for Pulsar Committer.
The transactions would
- * be removed until Flink triggered a checkpoint.
- */
- public List<PulsarCommittable> prepareCommit() {
- List<PulsarCommittable> committables = new
ArrayList<>(transactions.size());
- for (Map.Entry<String, Transaction> entry : transactions.entrySet()) {
- String topic = entry.getKey();
- Transaction transaction = entry.getValue();
- TxnID txnID = transaction.getTxnID();
-
- committables.add(new PulsarCommittable(txnID, topic));
+ /** Return the transaction for the current checkpoint and clear it from
the writer state. */
+ @Nullable
+ public TxnID prepareCommit() {
+ if (transaction == null) {
+ return null;
}
- transactions.clear();
- return committables;
+ TxnID txnID = transaction.getTxnID();
+ transaction = null;
+ return txnID;
}
/**
@@ -272,18 +262,14 @@ public class ProducerRegister implements Closeable {
}
}
- /**
- * Get the cached topic-related transaction. Or create a new transaction
after checkpointing.
- */
- private Transaction getOrCreateTransaction(String topic) throws
PulsarClientException {
- if (transactions.containsKey(topic)) {
- return transactions.get(topic);
+ /** Get the transaction for the current checkpoint, or create one after
checkpointing. */
+ private Transaction getOrCreateTransaction() throws PulsarClientException {
+ if (transaction != null) {
+ return transaction;
}
long timeoutMillis = sinkConfiguration.getTransactionTimeoutMillis();
- Transaction transaction = createTransaction(pulsarClient,
timeoutMillis);
- transactions.put(topic, transaction);
-
+ transaction = createTransaction(pulsarClient, timeoutMillis);
return transaction;
}
@@ -299,19 +285,16 @@ public class ProducerRegister implements Closeable {
}
}
- /** Abort the existed transactions. This method would be used when closing
PulsarWriter. */
+ /** Abort the outstanding transaction. This method is used when closing
PulsarWriter. */
private void abortTransactions() {
- if (coordinatorClient == null || transactions.isEmpty()) {
+ if (coordinatorClient == null || transaction == null) {
return;
}
+ TxnID txnID = transaction.getTxnID();
try (Closer closer = Closer.create()) {
- for (Transaction transaction : transactions.values()) {
- TxnID txnID = transaction.getTxnID();
- closer.register(() -> coordinatorClient.abort(txnID));
- }
-
- transactions.clear();
+ closer.register(() -> coordinatorClient.abort(txnID));
+ transaction = null;
} catch (IOException e) {
throw new FlinkRuntimeException(e);
}
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializerTest.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializerTest.java
index f13a4bf..a0fe60f 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializerTest.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializerTest.java
@@ -21,11 +21,19 @@ package org.apache.flink.connector.pulsar.sink.committer;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.junit.jupiter.api.Test;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
-import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Tests for serializing and deserializing {@link PulsarCommittable} with
{@link
@@ -36,18 +44,154 @@ class PulsarCommittableSerializerTest {
private static final PulsarCommittableSerializer INSTANCE = new
PulsarCommittableSerializer();
@Test
- void committableSerDe() throws IOException {
- String topic = randomAlphabetic(10);
+ void v2SerializeAndDeserializeWithMessages() throws IOException {
TxnID txnID =
new TxnID(
ThreadLocalRandom.current().nextLong(),
ThreadLocalRandom.current().nextLong());
+ Map<String, MessageIdPojo> messages = new HashMap<>();
+ messages.put("topic-a", new MessageIdPojo(1L, 2L, 3, 0, -1));
+ messages.put("topic-b", new MessageIdPojo(10L, 20L, 1, 0, -1));
- PulsarCommittable committable = new PulsarCommittable(txnID, topic);
+ PulsarCommittable committable = new PulsarCommittable(txnID, messages);
byte[] bytes = INSTANCE.serialize(committable);
PulsarCommittable committable1 =
INSTANCE.deserialize(INSTANCE.getVersion(), bytes);
assertThat(committable1).isEqualTo(committable);
}
+
+ @Test
+ void v2SerializeAndDeserializeWithoutMessages() throws IOException {
+ PulsarCommittable committable =
+ new PulsarCommittable(new TxnID(1L, 2L),
Collections.emptyMap());
+
+ byte[] bytes = INSTANCE.serialize(committable);
+ PulsarCommittable deserialized =
INSTANCE.deserialize(INSTANCE.getVersion(), bytes);
+
+ assertThat(deserialized).isEqualTo(committable);
+ assertThat(deserialized.getLatestPublishedMessages()).isEmpty();
+ }
+
+ @Test
+ void v1DeserializationIsCompatible() throws IOException {
+ TxnID txnID =
+ new TxnID(
+ ThreadLocalRandom.current().nextLong(),
+ ThreadLocalRandom.current().nextLong());
+ byte[] serialized;
+ try (ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream output = new DataOutputStream(bytes)) {
+ output.writeLong(txnID.getMostSigBits());
+ output.writeLong(txnID.getLeastSigBits());
+ output.writeUTF("Topic_Placeholder");
+ output.flush();
+ serialized = bytes.toByteArray();
+ }
+
+ PulsarCommittable committable = INSTANCE.deserialize(1, serialized);
+
+ assertThat(committable.getTxnID()).isEqualTo(txnID);
+ assertThat(committable.getLatestPublishedMessages()).isEmpty();
+
+ byte[] v2Serialized = INSTANCE.serialize(committable);
+ PulsarCommittable roundTripped =
INSTANCE.deserialize(INSTANCE.getVersion(), v2Serialized);
+ assertThat(roundTripped).isEqualTo(committable);
+ }
+
+ @Test
+ void v2StateIsReadableByV1Deserializer() throws IOException {
+ TxnID txnID = new TxnID(1L, 2L);
+ byte[] serialized =
+ INSTANCE.serialize(
+ new PulsarCommittable(
+ txnID, Collections.singletonMap("topic",
messageId())));
+
+ // Connector versions that only know V1 ignore the serializer version
and trailing bytes.
+ try (DataInputStream input = new DataInputStream(new
ByteArrayInputStream(serialized))) {
+ assertThat(new TxnID(input.readLong(),
input.readLong())).isEqualTo(txnID);
+
assertThat(input.readUTF()).isEqualTo("unknown_due_to_v2_serialization");
+ }
+ }
+
+ @Test
+ void unknownVersionFailsDeserialization() {
+ assertThatThrownBy(() -> INSTANCE.deserialize(3, new byte[0]))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Unsupported Pulsar committable
version");
+ }
+
+ @Test
+ void truncatedV1StateFailsDeserialization() {
+ assertThatThrownBy(() -> INSTANCE.deserialize(1, new byte[16]))
+ .isInstanceOf(IOException.class);
+ }
+
+ @Test
+ void emptyTopicInV1StateFailsDeserialization() throws IOException {
+ byte[] serialized;
+ try (ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream output = new DataOutputStream(bytes)) {
+ output.writeLong(1L);
+ output.writeLong(2L);
+ output.writeUTF("");
+ output.flush();
+ serialized = bytes.toByteArray();
+ }
+
+ assertThatThrownBy(() -> INSTANCE.deserialize(1, serialized))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("topic must not be empty");
+ }
+
+ @Test
+ void negativeMessageCountInV2StateFailsDeserialization() throws
IOException {
+ byte[] serialized;
+ try (ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream output = new DataOutputStream(bytes)) {
+ output.writeLong(1L);
+ output.writeLong(2L);
+ output.writeUTF("topic");
+ output.writeInt(-1);
+ output.flush();
+ serialized = bytes.toByteArray();
+ }
+
+ assertThatThrownBy(() -> INSTANCE.deserialize(2, serialized))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Negative number of published messages");
+ }
+
+ @Test
+ void oversizedMessageCountInV2StateFailsWithoutPreallocating() throws
IOException {
+ byte[] serialized;
+ try (ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ DataOutputStream output = new DataOutputStream(bytes)) {
+ output.writeLong(1L);
+ output.writeLong(2L);
+ output.writeUTF("topic");
+ output.writeInt(Integer.MAX_VALUE);
+ output.flush();
+ serialized = bytes.toByteArray();
+ }
+
+ assertThatThrownBy(() -> INSTANCE.deserialize(2, serialized))
+ .isInstanceOf(IOException.class);
+ }
+
+ @Test
+ void trailingBytesInV2StateFailDeserialization() throws IOException {
+ PulsarCommittable committable =
+ new PulsarCommittable(new TxnID(1L, 2L),
Collections.emptyMap());
+ byte[] serialized = INSTANCE.serialize(committable);
+ byte[] serializedWithTrailingByte = Arrays.copyOf(serialized,
serialized.length + 1);
+
+ assertThatThrownBy(() -> INSTANCE.deserialize(2,
serializedWithTrailingByte))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Trailing bytes");
+ }
+
+ private static MessageIdPojo messageId() {
+ return new MessageIdPojo(1L, 2L, 3, 0, -1);
+ }
}
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
index 31ffd1b..b72a115 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
@@ -127,6 +127,7 @@ class PulsarWriterTest extends PulsarTestSuiteBase {
assertThat(committables).hasSize(1);
PulsarCommittable committable =
committables.stream().findFirst().orElseThrow(IllegalArgumentException::new);
+ assertThat(committable.getLatestPublishedMessages()).hasSize(1);
TransactionCoordinatorClient coordinatorClient =
operator().coordinatorClient();
coordinatorClient.commit(committable.getTxnID());
}
diff --git
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegisterTest.java
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegisterTest.java
index 03fef2b..2e82606 100644
---
a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegisterTest.java
+++
b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegisterTest.java
@@ -21,7 +21,6 @@ package org.apache.flink.connector.pulsar.sink.writer.topic;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.pulsar.common.crypto.PulsarCrypto;
-import org.apache.flink.connector.pulsar.sink.committer.PulsarCommittable;
import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
import org.apache.flink.connector.pulsar.testutils.PulsarTestSuiteBase;
@@ -35,7 +34,6 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
-import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
@@ -65,12 +63,9 @@ class ProducerRegisterTest extends PulsarTestSuiteBase {
register.createMessageBuilder(topic,
Schema.STRING).value(message).send();
if (deliveryGuarantee == EXACTLY_ONCE) {
- List<PulsarCommittable> committables = register.prepareCommit();
- for (PulsarCommittable committable : committables) {
- TxnID txnID = committable.getTxnID();
- TransactionCoordinatorClient coordinatorClient =
operator().coordinatorClient();
- coordinatorClient.commit(txnID);
- }
+ TxnID txnID = register.prepareCommit();
+ TransactionCoordinatorClient coordinatorClient =
operator().coordinatorClient();
+ coordinatorClient.commit(txnID);
}
Message<String> receiveMessage = operator().receiveMessage(topic,
Schema.STRING);
@@ -95,8 +90,33 @@ class ProducerRegisterTest extends PulsarTestSuiteBase {
String message = randomAlphabetic(10);
register.createMessageBuilder(topic,
Schema.STRING).value(message).sendAsync();
- List<PulsarCommittable> committables = register.prepareCommit();
- assertThat(committables).isEmpty();
+ assertThat(register.prepareCommit()).isNull();
+ }
+
+ @Test
+ void exactlyOnceUsesOneTransactionAcrossTopicsPerCheckpoint() throws
Exception {
+ String firstTopic = randomAlphabetic(10);
+ String secondTopic = randomAlphabetic(10);
+ operator().createTopic(firstTopic, 8);
+ operator().createTopic(secondTopic, 8);
+
+ SinkConfiguration configuration =
+ new
SinkConfiguration(operator().sinkConfig(DeliveryGuarantee.EXACTLY_ONCE));
+ ProducerRegister register =
+ new ProducerRegister(
+ configuration, PulsarCrypto.disabled(),
createSinkWriterMetricGroup());
+
+ register.createMessageBuilder(firstTopic,
Schema.STRING).value("first").send();
+ register.createMessageBuilder(secondTopic,
Schema.STRING).value("second").send();
+
+ TxnID txnID = register.prepareCommit();
+ assertThat(txnID).isNotNull();
+ operator().coordinatorClient().commit(txnID);
+
+ assertThat(operator().receiveMessage(firstTopic,
Schema.STRING).getValue())
+ .isEqualTo("first");
+ assertThat(operator().receiveMessage(secondTopic,
Schema.STRING).getValue())
+ .isEqualTo("second");
}
@Test
