ferenc-csaky commented on code in PR #130:
URL:
https://github.com/apache/flink-connector-pulsar/pull/130#discussion_r4313883226
##########
flink-sql-connector-pulsar/pom.xml:
##########
@@ -72,6 +80,8 @@ under the License.
<artifactSet>
<includes>
<include>com.fasterxml.jackson.core:jackson-annotations</include>
+
<include>com.fasterxml.jackson.core:jackson-core</include>
+
<include>com.fasterxml.jackson.core:jackson-databind</include>
Review Comment:
If Jackson stays, these should be relocated, like `org.bouncycastle` and
`com.scurrilous` below. The SQL jar goes into `lib/`, and an unrelocated
`com.fasterxml.jackson.databind` there can clash with other jars that bring
their own Jackson (Hadoop, other formats or connectors). `jackson-annotations`
was already bundled unrelocated, but databind is a much bigger surface. This is
moot if V2 moves to a binary format.
##########
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:
`serialize` writes the domain `PulsarCommittable` (which contains Pulsar's
`TxnID`), but `deserialize` reads `PulsarCommittablePojo`. So the persisted
format comes from whatever getters `PulsarCommittable` and `TxnID` expose, not
from the POJOs. `getVersion()` already lands in the JSON as `"version"`. If a
Pulsar release renamed a `TxnID` getter, our checkpoint format would change
silently and old state would restore as `TxnID(0,0)`. Converting to the POJO
and serializing that would make the format explicit.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java:
##########
@@ -167,14 +174,25 @@ 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 {
+ // Safe cast: all MessageId implementations in
Pulsar implement
+ // MessageIdAdv.
+ MessageIdAdv messageIdAdv = (MessageIdAdv) id;
+ latestPublishedMessages.put(
+ topic,
+ new MessageIdPojo(
+ messageIdAdv.getLedgerId(),
+ messageIdAdv.getEntryId(),
+ messageIdAdv.getBatchSize(),
+ messageIdAdv.getBatchIndex(),
+ messageIdAdv.getPartitionIndex()));
LOG.debug("Sent message to Pulsar {} with message
id {}", topic, id);
}
+ pendingMessages.decrementAndGet();
Review Comment:
I assume the decrement moved to the end so the map update is visible before
`flush()` returns. But now anything that throws above it skips the decrement,
and `whenComplete` swallows the exception. That includes a rejected
`mailboxExecutor.execute` in the error branch. `flush()` then spins forever in
`while (pendingMessages.longValue() > 0)` and the checkpoint hangs without a
log line. Wrapping the callback body in `try { ... } finally {
pendingMessages.decrementAndGet(); }` keeps the ordering and avoids the hang.
Nit: the comment on L182 overstates things, since `MultiMessageIdImpl`
doesn't implement `MessageIdAdv`. The producer send path only returns
`MessageIdImpl` subclasses, so the cast is fine, but I'd reword the comment to
say that instead.
##########
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:
This map is filled on every successful send for every delivery guarantee,
but it's only read and cleared under `EXACTLY_ONCE`. For AT_LEAST_ONCE and
NONE, that costs an allocation and a map write per record, and the map keeps
growing with every topic ever written. I'd track only under `EXACTLY_ONCE`. You
could also store the `MessageIdAdv` here and convert it to `MessageIdPojo` once
per topic in `prepareCommit`.
##########
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:
Any version other than 1 falls through to V2. An unknown version, for
example state written by a newer connector, should fail with a clear
`IOException` rather than be parsed as V2. Similarly, if `txnID` is missing,
L85 throws a bare NPE. It would be better to check it and throw an
`IOException` that says the state is corrupt.
##########
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:
This now produces at most one committable, but it still returns a `List`,
and it builds the committable with the `@Deprecated` constructor that is
documented as serializer-only. `PulsarWriter` then calls `iterator().next()`
and rebuilds the committable with the message IDs. Returning a `@Nullable
TxnID` would be simpler and make the one-transaction invariant explicit.
Alternatively, `ProducerRegister` could also own the latest-message-ID map.
Then the transaction and its IDs would be collected and reset together, instead
of depending on the flush/prepareCommit ordering comment in `PulsarWriter`.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/MessageIdPojo.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.PublicEvolving;
+
+import java.util.Objects;
+
+/** A POJO representation of Pulsar message ID used for serialization. */
+@PublicEvolving
Review Comment:
`PulsarCommittable` is `@Internal`, and these POJOs exist only to serialize
state, so I'd make them (`MessageIdPojo`, `TxnIdPojo` and
`PulsarCommittablePojo`) `@Internal` too. As `@PublicEvolving` they become API
surface, which makes it harder to evolve the state format later.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegister.java:
##########
@@ -273,17 +267,17 @@ private void configPulsarCrypto(ProducerBuilder<?>
builder) {
}
/**
- * Get the cached topic-related transaction. Or create a new transaction
after checkpointing.
+ * Get the transaction for the current checkpoint, or create one after
checkpointing.
+ *
+ * @param topic Used only for the message builder; transactions may span
topics.
Review Comment:
`topic` isn't used in this method any more. I'd remove the parameter instead
of documenting it.
##########
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);
+ }
+
+ 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);
- String topic = in.readUTF();
- return new PulsarCommittable(txnID, topic);
+ return new PulsarCommittable(txnID);
}
}
+
+ private PulsarCommittable deserializeV2(byte[] serialized) throws
IOException {
+ PulsarCommittablePojo pulsarCommittablePojo =
+ OBJECT_MAPPER.readValue(serialized,
PulsarCommittablePojo.class);
+ TxnID txnID =
+ new TxnID(
+ pulsarCommittablePojo.getTxnID().getMostSigBits(),
+ pulsarCommittablePojo.getTxnID().getLeastSigBits());
+ return new PulsarCommittable(txnID,
pulsarCommittablePojo.getLatestPublishedMessages());
Review Comment:
This always creates a version-2 committable, and `PulsarCommittablePojo` has
no version field, so a committable restored from V1 state turns into V2 after
one round trip. For example: restore V1 state, the commit is retried through
`retryLater`, the committer state is snapshotted again as JSON, and it restores
as `version=2` with an empty map. The `"version":1` does end up in the JSON
(see the comment on L59), but the POJO drops it. If the follow-up uses the
version to choose a recovery path, it will treat legacy committables as V2
committables with no published messages. Could you persist the version in the
POJO? `v1DeserializationIsCompatible` only checks the first deserialization, so
a round-trip assertion there would cover this.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java:
##########
@@ -167,14 +174,25 @@ 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 {
+ // Safe cast: all MessageId implementations in
Pulsar implement
+ // MessageIdAdv.
+ MessageIdAdv messageIdAdv = (MessageIdAdv) id;
+ latestPublishedMessages.put(
Review Comment:
This keeps whichever callback finishes last, not the highest message ID. One
topic can have several producers in `ProducerRegister` (one per `SchemaHash`),
and their callbacks complete independently, so an older ID can overwrite a
newer one. Also, with broker dedup and a user-supplied `sequenceId` (L255), a
duplicate is acked with ledger/entry `-1`, which overwrites the real position.
The follow-up will rely on this value to decide between commit and abort, so
I'd keep the maximum per topic (`compute` with a `MessageIdAdv` comparison) and
skip negative IDs.
##########
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:
Most of these are already Jackson's defaults, or they cover features these
beans don't use (creators, subtypes, object IDs, unwrapping). Only
`FAIL_ON_UNKNOWN_PROPERTIES` actually changes anything. The resulting behaviour
is also too lenient for checkpoint state: trailing tokens are ignored and
missing fields default to 0 or null, so `{"txnID":{}} garbage` deserializes to
`TxnID(0,0)`, which the committer would then try to commit. I'd remove this
block (keeping only what's needed) and fail fast on corrupt state.
##########
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;
Review Comment:
This brings unshaded `com.fasterxml.jackson.databind` into the connector,
but `flink-connector-pulsar` doesn't declare it. Today it only arrives through
`flink-avro` → `avro`, which is `provided` and optional; `pulsar-client-all`
only brings `jackson-annotations`. A DataStream job using the plain connector
without its own jackson-databind will fail at submission with
`NoClassDefFoundError: com/fasterxml/jackson/databind/ObjectMapper`. That
happens for every delivery guarantee, because `getCommittableSerializer()` is
called during sink translation.
Unless there's a strong reason for JSON, I'd keep V2 binary with
`DataOutputStream` like V1: the txn longs, the map size, then `writeUTF(topic)`
plus the message ID fields for each entry. That removes the dependency and the
shading question. If you prefer JSON, Flink's shaded Jackson
(`org.apache.flink.util.jackson.JacksonMapperFactory`) is already on the
runtime classpath.
##########
flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/committer/PulsarCommittable.java:
##########
@@ -31,41 +33,58 @@ 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 = txnID;
+ this.latestPublishedMessages = latestPublishedMessages;
+ this.version = 2;
+ }
+
+ // This constructor should only be used by PulsarCommittableSerializer.
+ @Deprecated
+ public PulsarCommittable(TxnID txnID) {
this.txnID = txnID;
- this.topic = topic;
+ this.latestPublishedMessages = Collections.emptyMap();
+ this.version = 1;
}
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);
}
@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
Review Comment:
`MessageIdPojo` has no `toString()`, so this prints identity hashes like
`MessageIdPojo@1b639ba` in every committer error log, one per partition
written. Could you add a `toString()`?
##########
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:
For the release notes: the previous serializer ignores the version argument
and always reads two longs and a UTF string. So after a checkpoint or savepoint
taken with this change, downgrading the connector doesn't fail cleanly. It
reads the JSON bytes as a garbage `TxnID` and then fails in the committer, or
on a bogus UTF length. That can't be fixed here, but we should state that
downgrading after this upgrade isn't supported.
--
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]
