tisonkun commented on code in PR #20725:
URL: https://github.com/apache/flink/pull/20725#discussion_r961732654
##########
flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssigner.java:
##########
@@ -34,7 +33,7 @@
* readers and store all the state into checkpoint.
*/
@Internal
-public interface SplitAssigner extends Serializable {
Review Comment:
Why do you make this change?
##########
flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarUnorderedSourceReader.java:
##########
@@ -80,6 +81,38 @@ public PulsarUnorderedSourceReader(
this.transactionsOfFinishedSplits = Collections.synchronizedList(new
ArrayList<>());
}
+ @Override
+ public void start() {
+ this.started = true;
+ super.start();
Review Comment:
```suggestion
```
##########
flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumStateSerializer.java:
##########
@@ -54,57 +55,37 @@ private PulsarSourceEnumStateSerializer() {
@Override
public int getVersion() {
- // We use PulsarPartitionSplitSerializer's version because we use
reuse this class.
- return PulsarPartitionSplitSerializer.CURRENT_VERSION;
+ return CURRENT_VERSION;
}
@Override
public byte[] serialize(PulsarSourceEnumState obj) throws IOException {
- // VERSION 0 serialization
try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream out = new DataOutputStream(baos)) {
serializeSet(
out, obj.getAppendedPartitions(),
SPLIT_SERIALIZER::serializeTopicPartition);
- serializeSet(
Review Comment:
Shall we serialize dummy states for downgrading from version 1 to version 0?
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
