Savonitar commented on code in PR #130:
URL:
https://github.com/apache/flink-connector-pulsar/pull/130#discussion_r4346453700
##########
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:
Do we have a test for a checkpoint containing only duplicate sends?
##########
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:
Should we skip IDs with a negative ledger or entry ID before updating the
map?
##########
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:
equals/hashcode() got new field `version`, does it make sense to add it here
(to toString) as well?
##########
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:
Would it make sense to use the default capacity, or to bound the count by
`input.available()` ?
--
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]
