poorbarcode commented on code in PR #130:
URL:
https://github.com/apache/flink-connector-pulsar/pull/130#discussion_r4332553316
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializer.java:
##########
@@ -20,46 +20,70 @@
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
import org.apache.pulsar.client.api.transaction.TxnID;
import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
-import java.io.DataOutputStream;
import java.io.IOException;
/** 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;
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ static {
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_IGNORED_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_INVALID_SUBTYPE, false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_NULL_CREATOR_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_MISSING_CREATOR_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_UNRESOLVED_OBJECT_IDS,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_TRAILING_TOKENS, false);
+ OBJECT_MAPPER.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS,
false);
+
OBJECT_MAPPER.configure(SerializationFeature.FAIL_ON_UNWRAPPED_TYPE_IDENTIFIERS,
false);
+ }
@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)) {
- TxnID txnID = obj.getTxnID();
- out.writeLong(txnID.getMostSigBits());
- out.writeLong(txnID.getLeastSigBits());
- out.writeUTF(obj.getTopic());
- out.flush();
- return baos.toByteArray();
- }
+ return OBJECT_MAPPER.writeValueAsBytes(obj);
Review Comment:
The V2 binary serializer now writes the version, TxnID fields, map size,
topics, and message-ID fields explicitly. The JSON DTOs have been removed.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializer.java:
##########
@@ -20,46 +20,70 @@
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
import org.apache.pulsar.client.api.transaction.TxnID;
import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
-import java.io.DataOutputStream;
import java.io.IOException;
/** 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;
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ static {
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_IGNORED_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_INVALID_SUBTYPE, false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_NULL_CREATOR_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_MISSING_CREATOR_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_UNRESOLVED_OBJECT_IDS,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_TRAILING_TOKENS, false);
+ OBJECT_MAPPER.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS,
false);
+
OBJECT_MAPPER.configure(SerializationFeature.FAIL_ON_UNWRAPPED_TYPE_IDENTIFIERS,
false);
+ }
Review Comment:
Removed the Jackson configuration entirely with the binary serializer. The
decoder now rejects unsupported versions, negative map sizes, trailing bytes,
and truncated payloads; dedicated tests cover these cases.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializer.java:
##########
@@ -20,46 +20,70 @@
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
import org.apache.pulsar.client.api.transaction.TxnID;
import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
-import java.io.DataOutputStream;
import java.io.IOException;
/** 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;
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ static {
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_IGNORED_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_INVALID_SUBTYPE, false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_NULL_CREATOR_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_MISSING_CREATOR_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_UNRESOLVED_OBJECT_IDS,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_TRAILING_TOKENS, false);
+ OBJECT_MAPPER.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS,
false);
+
OBJECT_MAPPER.configure(SerializationFeature.FAIL_ON_UNWRAPPED_TYPE_IDENTIFIERS,
false);
+ }
@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)) {
- TxnID txnID = obj.getTxnID();
- out.writeLong(txnID.getMostSigBits());
- out.writeLong(txnID.getLeastSigBits());
- out.writeUTF(obj.getTopic());
- out.flush();
- return baos.toByteArray();
- }
+ return OBJECT_MAPPER.writeValueAsBytes(obj);
}
@Override
public PulsarCommittable deserialize(int version, byte[] serialized)
throws IOException {
+ if (version == VERSION_V1) {
+ return deserializeV1(serialized);
+ }
+ return deserializeV2(serialized);
Review Comment:
deserialize now switches explicitly on the outer serializer version and
rejects unknown versions with IOException. The binary decoder reads all
required TxnID primitives directly, so truncated state fails as IOException.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittableSerializer.java:
##########
@@ -20,46 +20,70 @@
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
import org.apache.pulsar.client.api.transaction.TxnID;
import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
-import java.io.DataOutputStream;
import java.io.IOException;
/** 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;
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ static {
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_IGNORED_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_INVALID_SUBTYPE, false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_NULL_CREATOR_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_MISSING_CREATOR_PROPERTIES,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_UNRESOLVED_OBJECT_IDS,
false);
+
OBJECT_MAPPER.configure(DeserializationFeature.FAIL_ON_TRAILING_TOKENS, false);
+ OBJECT_MAPPER.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS,
false);
+
OBJECT_MAPPER.configure(SerializationFeature.FAIL_ON_UNWRAPPED_TYPE_IDENTIFIERS,
false);
+ }
@Override
public int getVersion() {
- return CURRENT_VERSION;
+ return VERSION_V2;
Review Comment:
Added a Compatibility section to the PR description documenting that
checkpoint/savepoint state written with V2 cannot be restored after downgrading
to a connector that supports only V1.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegister.java:
##########
@@ -158,21 +157,16 @@ public <T> TypedMessageBuilder<T>
createMessageBuilder(String topic, @Nullable S
}
/**
- * Convert the transactions into a committable list for Pulsar Committer.
The transactions would
- * be removed until Flink triggered a checkpoint.
+ * Convert the transaction for the current checkpoint into a committable
for Pulsar Committer.
*/
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));
+ if (transaction == null) {
+ return Collections.emptyList();
}
- transactions.clear();
- return committables;
+ TxnID txnID = transaction.getTxnID();
+ transaction = null;
+ return Collections.singletonList(new PulsarCommittable(txnID));
}
Review Comment:
Changed ProducerRegister.prepareCommit() to return a nullable TxnID. This
makes the one-transaction invariant explicit and avoids constructing then
rebuilding an intermediate committable.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java:
##########
@@ -79,6 +84,8 @@ public class PulsarWriter<IN> implements
CommittingSinkWriter<IN, PulsarCommitta
private final ProducerRegister producerRegister;
private final MailboxExecutor mailboxExecutor;
private final AtomicLong pendingMessages;
+ private final ConcurrentHashMap<String, MessageIdPojo>
latestPublishedMessages =
+ new ConcurrentHashMap<>();
Review Comment:
Latest message IDs are now tracked only for EXACTLY_ONCE. The map keeps
MessageIdAdv values during writes and converts them to MessageIdPojo only once
in prepareCommit.
--
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]
