ferenc-csaky commented on code in PR #8: URL: https://github.com/apache/flink-connector-kudu/pull/8#discussion_r1971718961
########## flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumerator.java: ########## @@ -0,0 +1,230 @@ +/* + * 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.kudu.source.enumerator; + +import org.apache.flink.api.connector.source.SourceEvent; +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.connector.kudu.connector.KuduTableInfo; +import org.apache.flink.connector.kudu.connector.reader.KuduReaderConfig; +import org.apache.flink.connector.kudu.source.split.KuduSourceSplit; +import org.apache.flink.connector.kudu.source.split.SplitFinishedEvent; + +import org.apache.kudu.util.HybridTimeUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +import java.io.IOException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + +/** + * The Kudu source enumerator is responsible for periodically discovering and assigning new splits + * when possible. + * + * <p>To provide CDC-like functionality, the enumeration works as follows: Initially, we perform a + * snapshot read of the table and mark the snapshot time as t0. From that point onward, we perform + * differential scans in the time intervals t0 - t1, t1 - t2, and so on. + * + * <p>This approach means that new splits can only be enumerated once the current time range is + * fully processed. + * + * <p>The process is controlled as follows: + * + * <ul> + * <li>Once a set of splits is enumerated for a time range, we track: + * <ul> + * <li><b>Unassigned splits</b>: Discovered but not yet assigned to readers. + * <li><b>Pending splits</b>: Assigned but not yet fully processed. + * </ul> + * <li>A new set of splits is generated only when there are no remaining unassigned or pending + * splits. + * </ul> + */ +public class KuduSourceEnumerator + implements SplitEnumerator<KuduSourceSplit, KuduSourceEnumeratorState> { + private static final Logger LOG = LoggerFactory.getLogger(KuduSourceEnumerator.class); + + private final SplitEnumeratorContext<KuduSourceSplit> context; + private final List<Integer> readersAwaitingSplit; + private final List<KuduSourceSplit> unassigned; + private final List<KuduSourceSplit> pending; + private final Duration period; + private final KuduSplitGenerator splitGenerator; + private long lastEndTimestamp; + + public KuduSourceEnumerator( + KuduTableInfo tableInfo, + KuduReaderConfig readerConfig, + Duration period, + SplitEnumeratorContext<KuduSourceSplit> context, + KuduSourceEnumeratorState enumState) { + this.period = period; + this.context = context; + this.readersAwaitingSplit = new ArrayList<>(); + this.unassigned = enumState == null ? new ArrayList<>() : enumState.getUnassigned(); + this.pending = enumState == null ? new ArrayList<>() : enumState.getPending(); + this.splitGenerator = new KuduSplitGenerator(readerConfig, tableInfo); + this.lastEndTimestamp = enumState == null ? -1L : enumState.getLastEndTimestamp(); + } + + @Override + public void start() { + context.callAsync( + () -> this.enumerateNewSplits(() -> pending.isEmpty() && unassigned.isEmpty()), + this::assignSplits, + 0, + this.period.toMillis()); + } + + @Override + public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname) { + readersAwaitingSplit.add(subtaskId); + assignSplitsToReaders(); + } + + @Override + public void addSplitsBack(List<KuduSourceSplit> splits, int subtaskId) { + LOG.debug("File Source Enumerator adds splits back: {}", splits); + unassigned.addAll(splits); + if (context.registeredReaders().containsKey(subtaskId)) { + readersAwaitingSplit.add(subtaskId); + } + assignSplitsToReaders(); + } + + @Override + public void addReader(int subtaskId) { + // The source is purely lazy-pull-based, nothing to do upon registration + } + + @Override + public KuduSourceEnumeratorState snapshotState(long checkpointId) throws Exception { + return new KuduSourceEnumeratorState(lastEndTimestamp, unassigned, pending); + } + + @Override + public void close() throws IOException { + try { + splitGenerator.close(); + } catch (Exception e) { + throw new IOException("Error closing split generator", e); + } + } + + @Override + public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) { + if (sourceEvent instanceof SplitFinishedEvent) { + SplitFinishedEvent splitFinishedEvent = (SplitFinishedEvent) sourceEvent; + LOG.debug( + "Received SplitFinishedEvent from subtask {} for splits: {}", + subtaskId, + splitFinishedEvent.getFinishedSplits()); + pending.removeAll(splitFinishedEvent.getFinishedSplits()); + readersAwaitingSplit.add(subtaskId); + } + } + + // This function is invoked repeatedly according to this.period if there are no outstanding + // splits. + // Outstanding meaning that there are no pending splits, and no enumerated but not assigned + // splits for the + // current period. + private List<KuduSourceSplit> enumerateNewSplits(@Nonnull Supplier<Boolean> shouldGenerate) { + if (shouldGenerate.get() == Boolean.FALSE) { Review Comment: Can be simplified to `if (!shouldGenerate.get()) ...`. ########## flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumerator.java: ########## @@ -0,0 +1,230 @@ +/* + * 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.kudu.source.enumerator; + +import org.apache.flink.api.connector.source.SourceEvent; +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.connector.kudu.connector.KuduTableInfo; +import org.apache.flink.connector.kudu.connector.reader.KuduReaderConfig; +import org.apache.flink.connector.kudu.source.split.KuduSourceSplit; +import org.apache.flink.connector.kudu.source.split.SplitFinishedEvent; + +import org.apache.kudu.util.HybridTimeUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +import java.io.IOException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + +/** + * The Kudu source enumerator is responsible for periodically discovering and assigning new splits + * when possible. + * + * <p>To provide CDC-like functionality, the enumeration works as follows: Initially, we perform a + * snapshot read of the table and mark the snapshot time as t0. From that point onward, we perform + * differential scans in the time intervals t0 - t1, t1 - t2, and so on. + * + * <p>This approach means that new splits can only be enumerated once the current time range is + * fully processed. + * + * <p>The process is controlled as follows: + * + * <ul> + * <li>Once a set of splits is enumerated for a time range, we track: + * <ul> + * <li><b>Unassigned splits</b>: Discovered but not yet assigned to readers. + * <li><b>Pending splits</b>: Assigned but not yet fully processed. + * </ul> + * <li>A new set of splits is generated only when there are no remaining unassigned or pending + * splits. + * </ul> + */ +public class KuduSourceEnumerator + implements SplitEnumerator<KuduSourceSplit, KuduSourceEnumeratorState> { + private static final Logger LOG = LoggerFactory.getLogger(KuduSourceEnumerator.class); + + private final SplitEnumeratorContext<KuduSourceSplit> context; + private final List<Integer> readersAwaitingSplit; + private final List<KuduSourceSplit> unassigned; + private final List<KuduSourceSplit> pending; + private final Duration period; + private final KuduSplitGenerator splitGenerator; + private long lastEndTimestamp; + + public KuduSourceEnumerator( + KuduTableInfo tableInfo, + KuduReaderConfig readerConfig, + Duration period, + SplitEnumeratorContext<KuduSourceSplit> context, + KuduSourceEnumeratorState enumState) { + this.period = period; + this.context = context; Review Comment: Lets wrap these with `checkNotNull`. ########## flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/KuduSourceBuilder.java: ########## @@ -0,0 +1,68 @@ +/* + * 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.kudu.source; + +import org.apache.flink.connector.kudu.connector.KuduTableInfo; +import org.apache.flink.connector.kudu.connector.converter.RowResultConverter; +import org.apache.flink.connector.kudu.connector.reader.KuduReaderConfig; + +import java.time.Duration; + +import static org.apache.flink.util.Preconditions.checkArgument; + +/** + * Builder to construct {@link KuduSource}. + * + * @param <OUT> type of the output records read from Kudu + */ +public class KuduSourceBuilder<OUT> { + private KuduReaderConfig readerConfig; + private KuduTableInfo tableInfo; + private RowResultConverter<OUT> rowResultConverter; + private Duration period; + + public KuduSourceBuilder<OUT> setTableInfo(KuduTableInfo tableInfo) { + this.tableInfo = tableInfo; + return this; + } + + public KuduSourceBuilder<OUT> setReaderConfig(KuduReaderConfig readerConfig) { + this.readerConfig = readerConfig; + return this; + } + + public KuduSourceBuilder<OUT> setRowResultConverter( + RowResultConverter<OUT> rowResultConverter) { + this.rowResultConverter = rowResultConverter; + return this; + } + + public KuduSourceBuilder<OUT> setPeriod(Duration period) { + this.period = period; + return this; + } + + public KuduSource<OUT> build() { + checkArgument(tableInfo != null, "Table info must be provided."); + checkArgument(readerConfig != null, "Reader config must be provided."); + checkArgument(rowResultConverter != null, "RowResultConverter must be provided."); + checkArgument(period != null, "Period must be provided."); Review Comment: Can be simplified to `checkNotNull`. ########## flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumerator.java: ########## @@ -0,0 +1,230 @@ +/* + * 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.kudu.source.enumerator; + +import org.apache.flink.api.connector.source.SourceEvent; +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.connector.kudu.connector.KuduTableInfo; +import org.apache.flink.connector.kudu.connector.reader.KuduReaderConfig; +import org.apache.flink.connector.kudu.source.split.KuduSourceSplit; +import org.apache.flink.connector.kudu.source.split.SplitFinishedEvent; + +import org.apache.kudu.util.HybridTimeUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +import java.io.IOException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + +/** + * The Kudu source enumerator is responsible for periodically discovering and assigning new splits + * when possible. + * + * <p>To provide CDC-like functionality, the enumeration works as follows: Initially, we perform a + * snapshot read of the table and mark the snapshot time as t0. From that point onward, we perform + * differential scans in the time intervals t0 - t1, t1 - t2, and so on. + * + * <p>This approach means that new splits can only be enumerated once the current time range is + * fully processed. + * + * <p>The process is controlled as follows: + * + * <ul> + * <li>Once a set of splits is enumerated for a time range, we track: + * <ul> + * <li><b>Unassigned splits</b>: Discovered but not yet assigned to readers. + * <li><b>Pending splits</b>: Assigned but not yet fully processed. + * </ul> + * <li>A new set of splits is generated only when there are no remaining unassigned or pending + * splits. + * </ul> + */ +public class KuduSourceEnumerator + implements SplitEnumerator<KuduSourceSplit, KuduSourceEnumeratorState> { + private static final Logger LOG = LoggerFactory.getLogger(KuduSourceEnumerator.class); + + private final SplitEnumeratorContext<KuduSourceSplit> context; + private final List<Integer> readersAwaitingSplit; + private final List<KuduSourceSplit> unassigned; + private final List<KuduSourceSplit> pending; + private final Duration period; + private final KuduSplitGenerator splitGenerator; + private long lastEndTimestamp; + + public KuduSourceEnumerator( + KuduTableInfo tableInfo, + KuduReaderConfig readerConfig, + Duration period, + SplitEnumeratorContext<KuduSourceSplit> context, + KuduSourceEnumeratorState enumState) { + this.period = period; + this.context = context; + this.readersAwaitingSplit = new ArrayList<>(); + this.unassigned = enumState == null ? new ArrayList<>() : enumState.getUnassigned(); + this.pending = enumState == null ? new ArrayList<>() : enumState.getPending(); + this.splitGenerator = new KuduSplitGenerator(readerConfig, tableInfo); + this.lastEndTimestamp = enumState == null ? -1L : enumState.getLastEndTimestamp(); + } + + @Override + public void start() { + context.callAsync( + () -> this.enumerateNewSplits(() -> pending.isEmpty() && unassigned.isEmpty()), + this::assignSplits, + 0, + this.period.toMillis()); + } + + @Override + public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname) { + readersAwaitingSplit.add(subtaskId); + assignSplitsToReaders(); + } + + @Override + public void addSplitsBack(List<KuduSourceSplit> splits, int subtaskId) { + LOG.debug("File Source Enumerator adds splits back: {}", splits); + unassigned.addAll(splits); + if (context.registeredReaders().containsKey(subtaskId)) { + readersAwaitingSplit.add(subtaskId); + } + assignSplitsToReaders(); + } + + @Override + public void addReader(int subtaskId) { + // The source is purely lazy-pull-based, nothing to do upon registration + } + + @Override + public KuduSourceEnumeratorState snapshotState(long checkpointId) throws Exception { + return new KuduSourceEnumeratorState(lastEndTimestamp, unassigned, pending); + } + + @Override + public void close() throws IOException { + try { + splitGenerator.close(); + } catch (Exception e) { + throw new IOException("Error closing split generator", e); + } + } + + @Override + public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) { + if (sourceEvent instanceof SplitFinishedEvent) { + SplitFinishedEvent splitFinishedEvent = (SplitFinishedEvent) sourceEvent; + LOG.debug( + "Received SplitFinishedEvent from subtask {} for splits: {}", + subtaskId, + splitFinishedEvent.getFinishedSplits()); + pending.removeAll(splitFinishedEvent.getFinishedSplits()); + readersAwaitingSplit.add(subtaskId); + } + } + + // This function is invoked repeatedly according to this.period if there are no outstanding + // splits. + // Outstanding meaning that there are no pending splits, and no enumerated but not assigned + // splits for the + // current period. + private List<KuduSourceSplit> enumerateNewSplits(@Nonnull Supplier<Boolean> shouldGenerate) { Review Comment: In general I'd invert the annotation usage for null/nonnull, marking `@Nullable` that can be `null`, and no annotation should mean it should not be `null`. This means a lot less annotations in the end if we are consistent with it and mark method params and class level fields where it applies. ########## flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/split/KuduSourceSplit.java: ########## @@ -0,0 +1,67 @@ +/* + * 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.kudu.source.split; + +import org.apache.flink.api.connector.source.SourceSplit; + +import java.io.Serializable; +import java.util.Arrays; +import java.util.Base64; + +/** The Kudu source split that wraps a Kudu scan token. */ +public class KuduSourceSplit implements SourceSplit, Serializable { + + // Serialized byte[] is already compact and Flink-optimized, whereas the + // full KuduScanToken object might include unnecessary metadata or complex internal structures. Review Comment: Lets do a proper javadoc-style comment here. ########## flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumeratorStateSerializer.java: ########## @@ -0,0 +1,96 @@ +/* + * 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.kudu.source.enumerator; + +import org.apache.flink.connector.kudu.source.split.KuduSourceSplit; +import org.apache.flink.connector.kudu.source.split.KuduSourceSplitSerializer; +import org.apache.flink.core.io.SimpleVersionedSerializer; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +/** The class that serializes and deserializes {@link KuduSourceEnumeratorState}. */ +public class KuduSourceEnumeratorStateSerializer + implements SimpleVersionedSerializer<KuduSourceEnumeratorState> { + + private static final int VERSION = 1; Review Comment: Rename this to `CURRENT_VERSION`, so we can keep the older versions as constants as well and even implement serialize logic for different versions at the same time. ########## flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/KuduSource.java: ########## @@ -0,0 +1,119 @@ +/* + * 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.kudu.source; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.api.connector.source.Boundedness; +import org.apache.flink.api.connector.source.Source; +import org.apache.flink.api.connector.source.SourceReader; +import org.apache.flink.api.connector.source.SourceReaderContext; +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.connector.kudu.connector.KuduTableInfo; +import org.apache.flink.connector.kudu.connector.converter.RowResultConverter; +import org.apache.flink.connector.kudu.connector.reader.KuduReaderConfig; +import org.apache.flink.connector.kudu.source.enumerator.KuduSourceEnumerator; +import org.apache.flink.connector.kudu.source.enumerator.KuduSourceEnumeratorState; +import org.apache.flink.connector.kudu.source.enumerator.KuduSourceEnumeratorStateSerializer; +import org.apache.flink.connector.kudu.source.reader.KuduSourceReader; +import org.apache.flink.connector.kudu.source.reader.KuduSourceSplitReader; +import org.apache.flink.connector.kudu.source.split.KuduSourceSplit; +import org.apache.flink.connector.kudu.source.split.KuduSourceSplitSerializer; +import org.apache.flink.core.io.SimpleVersionedSerializer; + +import java.time.Duration; + +/** + * A continuous, unbounded Flink {@link Source} for reading data from Apache Kudu. It uses + * differential scanning to achieve a CDC-like behavior, continuously capturing changes. + * + * <p>Key components: + * + * <ul> + * <li>{@link KuduReaderConfig} - Configures the Kudu connection, including master addresses. + * <li>{@link KuduTableInfo} - Specifies the target Kudu table, including its name and schema + * details. + * <li>{@link RowResultConverter} - Converts Kudu's {@code RowResult} into the desired output type + * {@code OUT}. + * <li>{@link Duration} - Defines the polling interval, i.e., the time between consecutive scans. + * </ul> + * + * @param <OUT> The type of the records produced by this source. + */ +@PublicEvolving +public class KuduSource<OUT> implements Source<OUT, KuduSourceSplit, KuduSourceEnumeratorState> { + private final KuduReaderConfig readerConfig; + private final KuduTableInfo tableInfo; + private final RowResultConverter<OUT> rowResultConverter; + private final Duration period; + + private final Configuration configuration; + + KuduSource( + KuduReaderConfig readerConfig, + KuduTableInfo tableInfo, + RowResultConverter<OUT> rowResultConverter, + Duration period) { + this.tableInfo = tableInfo; + this.readerConfig = readerConfig; + this.rowResultConverter = rowResultConverter; + this.period = period; + this.configuration = new Configuration(); + } + + @Override + public Boundedness getBoundedness() { + return Boundedness.CONTINUOUS_UNBOUNDED; Review Comment: A Kudu table is a finite source, so we should definitely not hardcode this to `CONTINUOUS_UNBOUNDED`. There can be some use-cases where the periodic rediscovery can be useful and a job can be kept up and running in a streaming fashion but I believe that should be explicitly configured and by default the `KuduSource` should behave as a `BOUNDED` source. ########## flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumerator.java: ########## @@ -0,0 +1,230 @@ +/* + * 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.kudu.source.enumerator; + +import org.apache.flink.api.connector.source.SourceEvent; +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.connector.kudu.connector.KuduTableInfo; +import org.apache.flink.connector.kudu.connector.reader.KuduReaderConfig; +import org.apache.flink.connector.kudu.source.split.KuduSourceSplit; +import org.apache.flink.connector.kudu.source.split.SplitFinishedEvent; + +import org.apache.kudu.util.HybridTimeUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +import java.io.IOException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + +/** + * The Kudu source enumerator is responsible for periodically discovering and assigning new splits + * when possible. + * + * <p>To provide CDC-like functionality, the enumeration works as follows: Initially, we perform a + * snapshot read of the table and mark the snapshot time as t0. From that point onward, we perform + * differential scans in the time intervals t0 - t1, t1 - t2, and so on. + * + * <p>This approach means that new splits can only be enumerated once the current time range is + * fully processed. + * + * <p>The process is controlled as follows: + * + * <ul> + * <li>Once a set of splits is enumerated for a time range, we track: + * <ul> + * <li><b>Unassigned splits</b>: Discovered but not yet assigned to readers. + * <li><b>Pending splits</b>: Assigned but not yet fully processed. + * </ul> + * <li>A new set of splits is generated only when there are no remaining unassigned or pending + * splits. + * </ul> + */ +public class KuduSourceEnumerator + implements SplitEnumerator<KuduSourceSplit, KuduSourceEnumeratorState> { + private static final Logger LOG = LoggerFactory.getLogger(KuduSourceEnumerator.class); + + private final SplitEnumeratorContext<KuduSourceSplit> context; + private final List<Integer> readersAwaitingSplit; + private final List<KuduSourceSplit> unassigned; + private final List<KuduSourceSplit> pending; + private final Duration period; + private final KuduSplitGenerator splitGenerator; + private long lastEndTimestamp; + + public KuduSourceEnumerator( + KuduTableInfo tableInfo, + KuduReaderConfig readerConfig, + Duration period, + SplitEnumeratorContext<KuduSourceSplit> context, + KuduSourceEnumeratorState enumState) { Review Comment: I'd overload this ctor with another one that does not take `enumState`, so we can spare those ugly inline `if` statements: ```java public KuduSourceEnumerator( KuduTableInfo tableInfo, KuduReaderConfig readerConfig, Duration period, SplitEnumeratorContext<KuduSourceSplit> context) { this(tableInfo, readerConfig, period, context, KuduSourceEnumeratorState.empty()); } ``` Then we can also remove the `null` in `KuduSource#createEnumerator()`. ########## flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumeratorState.java: ########## @@ -0,0 +1,52 @@ +/* + * 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.kudu.source.enumerator; + +import org.apache.flink.connector.kudu.source.split.KuduSourceSplit; + +import java.io.Serializable; +import java.util.List; + +/** The class storing state information for {@link KuduSourceEnumerator}. */ +public class KuduSourceEnumeratorState implements Serializable { + private static final long serialVersionUID = 1L; + private final long lastEndTimestamp; + private final List<KuduSourceSplit> unassigned; + private final List<KuduSourceSplit> pending; + + public KuduSourceEnumeratorState( Review Comment: I suggest to add a `static` factory method `empty()` that returns a properly initialized empty instance of the class, so it won't be necessary anywhere outside this class. ########## flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumerator.java: ########## @@ -0,0 +1,230 @@ +/* + * 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.kudu.source.enumerator; + +import org.apache.flink.api.connector.source.SourceEvent; +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.connector.kudu.connector.KuduTableInfo; +import org.apache.flink.connector.kudu.connector.reader.KuduReaderConfig; +import org.apache.flink.connector.kudu.source.split.KuduSourceSplit; +import org.apache.flink.connector.kudu.source.split.SplitFinishedEvent; + +import org.apache.kudu.util.HybridTimeUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +import java.io.IOException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + +/** + * The Kudu source enumerator is responsible for periodically discovering and assigning new splits + * when possible. + * + * <p>To provide CDC-like functionality, the enumeration works as follows: Initially, we perform a + * snapshot read of the table and mark the snapshot time as t0. From that point onward, we perform + * differential scans in the time intervals t0 - t1, t1 - t2, and so on. + * + * <p>This approach means that new splits can only be enumerated once the current time range is + * fully processed. + * + * <p>The process is controlled as follows: + * + * <ul> + * <li>Once a set of splits is enumerated for a time range, we track: + * <ul> + * <li><b>Unassigned splits</b>: Discovered but not yet assigned to readers. + * <li><b>Pending splits</b>: Assigned but not yet fully processed. + * </ul> + * <li>A new set of splits is generated only when there are no remaining unassigned or pending + * splits. + * </ul> + */ +public class KuduSourceEnumerator + implements SplitEnumerator<KuduSourceSplit, KuduSourceEnumeratorState> { + private static final Logger LOG = LoggerFactory.getLogger(KuduSourceEnumerator.class); + + private final SplitEnumeratorContext<KuduSourceSplit> context; + private final List<Integer> readersAwaitingSplit; + private final List<KuduSourceSplit> unassigned; + private final List<KuduSourceSplit> pending; + private final Duration period; + private final KuduSplitGenerator splitGenerator; + private long lastEndTimestamp; + + public KuduSourceEnumerator( + KuduTableInfo tableInfo, + KuduReaderConfig readerConfig, + Duration period, + SplitEnumeratorContext<KuduSourceSplit> context, + KuduSourceEnumeratorState enumState) { + this.period = period; + this.context = context; + this.readersAwaitingSplit = new ArrayList<>(); + this.unassigned = enumState == null ? new ArrayList<>() : enumState.getUnassigned(); + this.pending = enumState == null ? new ArrayList<>() : enumState.getPending(); + this.splitGenerator = new KuduSplitGenerator(readerConfig, tableInfo); + this.lastEndTimestamp = enumState == null ? -1L : enumState.getLastEndTimestamp(); + } + + @Override + public void start() { + context.callAsync( + () -> this.enumerateNewSplits(() -> pending.isEmpty() && unassigned.isEmpty()), + this::assignSplits, + 0, + this.period.toMillis()); + } + + @Override + public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname) { + readersAwaitingSplit.add(subtaskId); + assignSplitsToReaders(); + } + + @Override + public void addSplitsBack(List<KuduSourceSplit> splits, int subtaskId) { + LOG.debug("File Source Enumerator adds splits back: {}", splits); + unassigned.addAll(splits); + if (context.registeredReaders().containsKey(subtaskId)) { + readersAwaitingSplit.add(subtaskId); + } + assignSplitsToReaders(); + } + + @Override + public void addReader(int subtaskId) { + // The source is purely lazy-pull-based, nothing to do upon registration + } + + @Override + public KuduSourceEnumeratorState snapshotState(long checkpointId) throws Exception { + return new KuduSourceEnumeratorState(lastEndTimestamp, unassigned, pending); + } + + @Override + public void close() throws IOException { + try { + splitGenerator.close(); + } catch (Exception e) { + throw new IOException("Error closing split generator", e); + } + } + + @Override + public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) { + if (sourceEvent instanceof SplitFinishedEvent) { + SplitFinishedEvent splitFinishedEvent = (SplitFinishedEvent) sourceEvent; + LOG.debug( + "Received SplitFinishedEvent from subtask {} for splits: {}", + subtaskId, + splitFinishedEvent.getFinishedSplits()); + pending.removeAll(splitFinishedEvent.getFinishedSplits()); + readersAwaitingSplit.add(subtaskId); + } + } + + // This function is invoked repeatedly according to this.period if there are no outstanding + // splits. + // Outstanding meaning that there are no pending splits, and no enumerated but not assigned + // splits for the + // current period. + private List<KuduSourceSplit> enumerateNewSplits(@Nonnull Supplier<Boolean> shouldGenerate) { + if (shouldGenerate.get() == Boolean.FALSE) { + return null; + } + List<KuduSourceSplit> newSplits; + + if (isFirstSplitGeneration()) { + this.lastEndTimestamp = getCurrentHybridTime(); + newSplits = this.splitGenerator.generateFullScanSplits(this.lastEndTimestamp); + } else { + long startHT = this.lastEndTimestamp; + long endHT = getCurrentHybridTime(); + newSplits = this.splitGenerator.generateIncrementalSplits(startHT, endHT); + this.lastEndTimestamp = endHT; Review Comment: Lets be consistent with `this.` usage. Either use it for every class level field call, or omit everywhere outside the ctor. Personally I prefer the latter, since the IDE colors everything properly IMO it is just noise. Same applies for every newly added class. ########## flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumerator.java: ########## @@ -0,0 +1,230 @@ +/* + * 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.kudu.source.enumerator; + +import org.apache.flink.api.connector.source.SourceEvent; +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.connector.kudu.connector.KuduTableInfo; +import org.apache.flink.connector.kudu.connector.reader.KuduReaderConfig; +import org.apache.flink.connector.kudu.source.split.KuduSourceSplit; +import org.apache.flink.connector.kudu.source.split.SplitFinishedEvent; + +import org.apache.kudu.util.HybridTimeUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +import java.io.IOException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + +/** + * The Kudu source enumerator is responsible for periodically discovering and assigning new splits + * when possible. + * + * <p>To provide CDC-like functionality, the enumeration works as follows: Initially, we perform a + * snapshot read of the table and mark the snapshot time as t0. From that point onward, we perform + * differential scans in the time intervals t0 - t1, t1 - t2, and so on. + * + * <p>This approach means that new splits can only be enumerated once the current time range is + * fully processed. + * + * <p>The process is controlled as follows: + * + * <ul> + * <li>Once a set of splits is enumerated for a time range, we track: + * <ul> + * <li><b>Unassigned splits</b>: Discovered but not yet assigned to readers. + * <li><b>Pending splits</b>: Assigned but not yet fully processed. + * </ul> + * <li>A new set of splits is generated only when there are no remaining unassigned or pending + * splits. + * </ul> + */ +public class KuduSourceEnumerator + implements SplitEnumerator<KuduSourceSplit, KuduSourceEnumeratorState> { + private static final Logger LOG = LoggerFactory.getLogger(KuduSourceEnumerator.class); + + private final SplitEnumeratorContext<KuduSourceSplit> context; + private final List<Integer> readersAwaitingSplit; + private final List<KuduSourceSplit> unassigned; + private final List<KuduSourceSplit> pending; + private final Duration period; + private final KuduSplitGenerator splitGenerator; + private long lastEndTimestamp; + + public KuduSourceEnumerator( + KuduTableInfo tableInfo, + KuduReaderConfig readerConfig, + Duration period, + SplitEnumeratorContext<KuduSourceSplit> context, + KuduSourceEnumeratorState enumState) { + this.period = period; + this.context = context; + this.readersAwaitingSplit = new ArrayList<>(); + this.unassigned = enumState == null ? new ArrayList<>() : enumState.getUnassigned(); + this.pending = enumState == null ? new ArrayList<>() : enumState.getPending(); + this.splitGenerator = new KuduSplitGenerator(readerConfig, tableInfo); + this.lastEndTimestamp = enumState == null ? -1L : enumState.getLastEndTimestamp(); + } + + @Override + public void start() { + context.callAsync( + () -> this.enumerateNewSplits(() -> pending.isEmpty() && unassigned.isEmpty()), + this::assignSplits, + 0, + this.period.toMillis()); + } + + @Override + public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname) { + readersAwaitingSplit.add(subtaskId); + assignSplitsToReaders(); + } + + @Override + public void addSplitsBack(List<KuduSourceSplit> splits, int subtaskId) { + LOG.debug("File Source Enumerator adds splits back: {}", splits); + unassigned.addAll(splits); + if (context.registeredReaders().containsKey(subtaskId)) { + readersAwaitingSplit.add(subtaskId); + } + assignSplitsToReaders(); + } + + @Override + public void addReader(int subtaskId) { + // The source is purely lazy-pull-based, nothing to do upon registration + } + + @Override + public KuduSourceEnumeratorState snapshotState(long checkpointId) throws Exception { + return new KuduSourceEnumeratorState(lastEndTimestamp, unassigned, pending); + } + + @Override + public void close() throws IOException { + try { + splitGenerator.close(); + } catch (Exception e) { + throw new IOException("Error closing split generator", e); + } + } + + @Override + public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) { + if (sourceEvent instanceof SplitFinishedEvent) { + SplitFinishedEvent splitFinishedEvent = (SplitFinishedEvent) sourceEvent; + LOG.debug( + "Received SplitFinishedEvent from subtask {} for splits: {}", + subtaskId, + splitFinishedEvent.getFinishedSplits()); + pending.removeAll(splitFinishedEvent.getFinishedSplits()); + readersAwaitingSplit.add(subtaskId); + } + } + + // This function is invoked repeatedly according to this.period if there are no outstanding + // splits. + // Outstanding meaning that there are no pending splits, and no enumerated but not assigned + // splits for the + // current period. + private List<KuduSourceSplit> enumerateNewSplits(@Nonnull Supplier<Boolean> shouldGenerate) { + if (shouldGenerate.get() == Boolean.FALSE) { + return null; + } + List<KuduSourceSplit> newSplits; + + if (isFirstSplitGeneration()) { + this.lastEndTimestamp = getCurrentHybridTime(); + newSplits = this.splitGenerator.generateFullScanSplits(this.lastEndTimestamp); + } else { + long startHT = this.lastEndTimestamp; + long endHT = getCurrentHybridTime(); + newSplits = this.splitGenerator.generateIncrementalSplits(startHT, endHT); + this.lastEndTimestamp = endHT; + } + + return newSplits; + } + + private void assignSplits(List<KuduSourceSplit> splits, Throwable error) { + if (error != null) { + LOG.error("Failed to enumerate Kudu splits.", error); + return; + } + + if (splits != null) { + this.unassigned.addAll(splits); + } + assignSplitsToReaders(); + } + + private void assignSplitsToReaders() { + final Iterator<Integer> awaitingSubtasks = readersAwaitingSplit.iterator(); + + while (awaitingSubtasks.hasNext()) { + final int awaitingSubtask = awaitingSubtasks.next(); + + // If the reader that requested another split has failed in the meantime, remove + // it from the list of waiting readers + if (!context.registeredReaders().containsKey(awaitingSubtask)) { + awaitingSubtasks.remove(); + continue; + } + + final Optional<KuduSourceSplit> nextSplit = getNextSplit(); + if (nextSplit.isPresent()) { + context.assignSplit(nextSplit.get(), awaitingSubtask); + awaitingSubtasks.remove(); + pending.add(nextSplit.get()); + } else { + break; + } + } + } + + private Optional<KuduSourceSplit> getNextSplit() { + if (unassigned.isEmpty()) { + return Optional.empty(); + } + Iterator<KuduSourceSplit> iterator = unassigned.iterator(); + KuduSourceSplit next = iterator.next(); + iterator.remove(); + return Optional.of(next); + } + + // Helper method to get the current Hybrid Time + private long getCurrentHybridTime() { + long fromMicros = TimeUnit.MILLISECONDS.toMicros(System.currentTimeMillis()); + return HybridTimeUtil.physicalAndLogicalToHTTimestamp(fromMicros, 0) + 1; Review Comment: I think this can be simplified to: ```java return HybridTimeUtil.clockTimestampToHTTimestamp(System.currentTimeMillis(), TimeUnit.MILLISECONDS); ``` So we can spare the conversion. What's the purpose of `+1`? ########## flink-connector-kudu/pom.xml: ########## @@ -39,6 +39,16 @@ under the License. <artifactId>flink-clients</artifactId> </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-base</artifactId> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-test-utils</artifactId> Review Comment: Both this and `flink-test-utils` are only required for tests, so they should be added to `test` scope only and I'd also move them down right above `kudu-test-utils` to keep `test` scoped deps together. ########## flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/connector/converter/RowResultRowConverter.java: ########## @@ -31,6 +32,9 @@ public Row convert(RowResult row) { Schema schema = row.getColumnProjection(); Row values = new Row(schema.getColumnCount()); + if (row.hasIsDeleted() && row.isDeleted()) { + values.setKind(RowKind.DELETE); + } Review Comment: Is this a bug currently? I guess this would be required for the currently existing `InputFormat` as well. ########## flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumerator.java: ########## @@ -0,0 +1,230 @@ +/* + * 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.kudu.source.enumerator; + +import org.apache.flink.api.connector.source.SourceEvent; +import org.apache.flink.api.connector.source.SplitEnumerator; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.connector.kudu.connector.KuduTableInfo; +import org.apache.flink.connector.kudu.connector.reader.KuduReaderConfig; +import org.apache.flink.connector.kudu.source.split.KuduSourceSplit; +import org.apache.flink.connector.kudu.source.split.SplitFinishedEvent; + +import org.apache.kudu.util.HybridTimeUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +import java.io.IOException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + +/** + * The Kudu source enumerator is responsible for periodically discovering and assigning new splits + * when possible. + * + * <p>To provide CDC-like functionality, the enumeration works as follows: Initially, we perform a + * snapshot read of the table and mark the snapshot time as t0. From that point onward, we perform + * differential scans in the time intervals t0 - t1, t1 - t2, and so on. + * + * <p>This approach means that new splits can only be enumerated once the current time range is + * fully processed. + * + * <p>The process is controlled as follows: + * + * <ul> + * <li>Once a set of splits is enumerated for a time range, we track: + * <ul> + * <li><b>Unassigned splits</b>: Discovered but not yet assigned to readers. + * <li><b>Pending splits</b>: Assigned but not yet fully processed. + * </ul> + * <li>A new set of splits is generated only when there are no remaining unassigned or pending + * splits. + * </ul> + */ +public class KuduSourceEnumerator + implements SplitEnumerator<KuduSourceSplit, KuduSourceEnumeratorState> { + private static final Logger LOG = LoggerFactory.getLogger(KuduSourceEnumerator.class); + + private final SplitEnumeratorContext<KuduSourceSplit> context; + private final List<Integer> readersAwaitingSplit; + private final List<KuduSourceSplit> unassigned; + private final List<KuduSourceSplit> pending; + private final Duration period; + private final KuduSplitGenerator splitGenerator; + private long lastEndTimestamp; Review Comment: nit: I'd reorder these fields and put the `KuduSourceEnumeratorState` related fields to the end in 1 group. ########## flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/split/KuduSourceSplitSerializer.java: ########## @@ -0,0 +1,54 @@ +/* + * 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.kudu.source.split; + +import org.apache.flink.core.io.SimpleVersionedSerializer; + +import java.io.IOException; + +/** The class that serializes and deserializes {@link KuduSourceSplit}. */ +public class KuduSourceSplitSerializer implements SimpleVersionedSerializer<KuduSourceSplit> { + + private static final int VERSION = 1; // Versioning for future changes Review Comment: Lets rename this to `CURRENT_VERSION`. And the comment can be removed, this is trivial enough. -- 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]
