poorbarcode commented on code in PR #130:
URL:
https://github.com/apache/flink-connector-pulsar/pull/130#discussion_r4347589174
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java:
##########
@@ -167,18 +178,38 @@ public void write(IN element, Context context) throws
IOException, InterruptedEx
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
Review Comment:
The successful producer-send callback only provides IDs for actually
published messages, whose ledger and entry IDs are non-negative. Therefore no
additional negative-ID filtering is needed here.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java:
##########
@@ -167,18 +178,38 @@ public void write(IN element, Context context) throws
IOException, InterruptedEx
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) {
Review Comment:
No additional duplicate-send test was added because the successful
producer-send callback is expected to return a real, non-negative ID for every
published message. Consequently, a checkpoint containing only duplicate sends
does not create a distinct sentinel-ID state to validate here.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializer.java:
##########
@@ -27,39 +27,107 @@
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}. */
public class PulsarCommittableSerializer implements
SimpleVersionedSerializer<PulsarCommittable> {
- private static final int CURRENT_VERSION = 1;
+ private static final int VERSION_V1 = 1;
+ private static final int VERSION_V2 = 2;
@Override
public int getVersion() {
- return CURRENT_VERSION;
+ return VERSION_V2;
}
@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)) {
+ output.writeInt(obj.getVersion());
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());
+
+ 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 {
+ switch (version) {
+ case VERSION_V1:
+ return deserializeV1(serialized);
+ case VERSION_V2:
+ return deserializeV2(serialized);
+ default:
+ throw new IOException("Unsupported Pulsar committable version:
" + version);
+ }
+ }
+
+ private PulsarCommittable deserializeV1(byte[] serialized) throws
IOException {
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 state also stores a topic. A transaction can now span
topics, so the field is
+ // not restored into the committable. Still read and validate it
to consume the
+ // complete V1 payload and reject corrupted legacy state instead
of silently
+ // accepting it.
String topic = in.readUTF();
- return new PulsarCommittable(txnID, topic);
+ if (topic.isEmpty()) {
+ throw new IOException("Corrupt V1 committable: topic must not
be empty.");
+ }
+ TxnID txnID = new TxnID(mostSigBits, leastSigBits);
+ return new PulsarCommittable(txnID, Collections.emptyMap(),
VERSION_V1);
+ }
+ }
+
+ private PulsarCommittable deserializeV2(byte[] serialized) throws
IOException {
+ try (ByteArrayInputStream bytes = new ByteArrayInputStream(serialized);
+ DataInputStream input = new DataInputStream(bytes)) {
+ int committableVersion = input.readInt();
+ if (committableVersion != VERSION_V1 && committableVersion !=
VERSION_V2) {
+ throw new IOException(
+ "Unsupported Pulsar committable state version: " +
committableVersion);
+ }
+
+ TxnID txnID = new TxnID(input.readLong(), input.readLong());
+ int messageCount = input.readInt();
+ if (messageCount < 0) {
+ throw new IOException("Negative number of published messages:
" + messageCount);
+ }
+
+ Map<String, MessageIdPojo> messages = new HashMap<>(messageCount);
Review Comment:
Changed this to use the default HashMap capacity. Also added a test with
Integer.MAX_VALUE as the serialized count; it now fails with IOException when
the payload ends, without preallocating the requested capacity.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittable.java:
##########
@@ -31,41 +32,56 @@ 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) {
+ private final int version;
+
+ public PulsarCommittable(TxnID txnID, Map<String, MessageIdPojo>
latestPublishedMessages) {
+ this(txnID, latestPublishedMessages, 2);
+ }
+
+ PulsarCommittable(
+ TxnID txnID, Map<String, MessageIdPojo> latestPublishedMessages,
int version) {
this.txnID = txnID;
- this.topic = topic;
+ this.latestPublishedMessages = latestPublishedMessages;
+ this.version = version;
}
public TxnID getTxnID() {
return txnID;
}
- public String getTopic() {
- return topic;
+ public Map<String, MessageIdPojo> getLatestPublishedMessages() {
+ return latestPublishedMessages;
+ }
+
+ public int getVersion() {
+ return version;
}
@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)
+ && version == that.version;
}
@Override
public int hashCode() {
- return Objects.hash(txnID, topic);
+ return Objects.hash(txnID, latestPublishedMessages, version);
}
@Override
public String toString() {
- return "PulsarCommittable{" + "txnID=" + txnID + ", topic='" + topic +
'\'' + '}';
+ return "PulsarCommittable{"
Review Comment:
Added version to PulsarCommittable.toString() and added a test that verifies
it is included.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
