This is an automated email from the ASF dual-hosted git repository.
fcsaky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kudu.git
The following commit(s) were added to refs/heads/main by this push:
new b823ed6 [FLINK-36855] Source API implementation
b823ed6 is described below
commit b823ed696e74b1efadc54c0cdbffc5c989aba30e
Author: martongreber <[email protected]>
AuthorDate: Fri Mar 14 15:52:27 2025 +0100
[FLINK-36855] Source API implementation
---
flink-connector-kudu/pom.xml | 17 ++
.../kudu/connector/reader/KuduReaderConfig.java | 115 ++++++++-
.../flink/connector/kudu/source/KuduSource.java | 133 +++++++++++
.../connector/kudu/source/KuduSourceBuilder.java | 80 +++++++
.../source/enumerator/KuduSourceEnumerator.java | 260 +++++++++++++++++++++
.../enumerator/KuduSourceEnumeratorState.java | 57 +++++
.../KuduSourceEnumeratorStateSerializer.java | 96 ++++++++
.../kudu/source/reader/KuduRecordEmitter.java | 43 ++++
.../kudu/source/reader/KuduSourceReader.java | 73 ++++++
.../kudu/source/reader/KuduSourceSplitReader.java | 105 +++++++++
.../kudu/source/split/KuduSourceSplit.java | 70 ++++++
.../source/split/KuduSourceSplitSerializer.java | 54 +++++
.../kudu/source/split/SplitFinishedEvent.java | 42 ++++
.../kudu/source/utils/KuduSourceUtils.java | 49 ++++
.../kudu/source/utils/KuduSplitGenerator.java | 122 ++++++++++
.../connector/kudu/connector/KuduTestBase.java | 9 +-
.../connector/kudu/source/KuduSourceITCase.java | 132 +++++++++++
.../connector/kudu/source/KuduSourceTest.java | 81 +++++++
.../connector/kudu/source/KuduSourceTestBase.java | 126 ++++++++++
.../KuduSourceEnumeratorStateSerializerTest.java | 85 +++++++
.../enumerator/KuduSourceEnumeratorTest.java | 138 +++++++++++
.../source/enumerator/KuduSplitGeneratorTest.java | 154 ++++++++++++
.../source/reader/KuduSourceSplitReaderTest.java | 80 +++++++
.../split/KuduSourceSplitSerializerTest.java | 60 +++++
pom.xml | 32 +++
25 files changed, 2205 insertions(+), 8 deletions(-)
diff --git a/flink-connector-kudu/pom.xml b/flink-connector-kudu/pom.xml
index 22817ff..6be5785 100644
--- a/flink-connector-kudu/pom.xml
+++ b/flink-connector-kudu/pom.xml
@@ -39,6 +39,11 @@ 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-streaming-java</artifactId>
@@ -64,11 +69,23 @@ under the License.
<artifactId>flink-table-runtime</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils</artifactId>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.kudu</groupId>
<artifactId>kudu-client</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-test-utils</artifactId>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.kudu</groupId>
<artifactId>kudu-test-utils</artifactId>
diff --git
a/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/connector/reader/KuduReaderConfig.java
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/connector/reader/KuduReaderConfig.java
index 71d56de..59e0bc6 100644
---
a/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/connector/reader/KuduReaderConfig.java
+++
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/connector/reader/KuduReaderConfig.java
@@ -21,6 +21,7 @@ import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.connector.kudu.format.KuduRowInputFormat;
import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.kudu.client.ReplicaSelection;
import java.io.Serializable;
@@ -35,11 +36,34 @@ public class KuduReaderConfig implements Serializable {
private final String masters;
private final int rowLimit;
-
- private KuduReaderConfig(String masters, int rowLimit) {
+ private final long splitSizeBytes;
+ private final int batchSizeBytes;
+ private final long scanRequestTimeout;
+ private final boolean prefetching;
+ private final long keepAlivePeriodMs;
+ private final ReplicaSelection replicaSelection;
+
+ private KuduReaderConfig(
+ String masters,
+ int rowLimit,
+ long splitSizeBytes,
+ int batchSizeBytes,
+ long scanRequestTimeout,
+ boolean prefetching,
+ long keepAlivePeriodMs,
+ ReplicaSelection replicaSelection) {
this.masters = checkNotNull(masters, "Kudu masters cannot be null");
this.rowLimit = checkNotNull(rowLimit, "Kudu rowLimit cannot be null");
+ this.splitSizeBytes = checkNotNull(splitSizeBytes, "Kudu split size
cannot be null");
+ this.batchSizeBytes = checkNotNull(batchSizeBytes, "Kudu batch size
cannot be null");
+ this.scanRequestTimeout =
+ checkNotNull(scanRequestTimeout, "Kudu scan request timeout
cannot be null");
+ this.prefetching = checkNotNull(prefetching, "Kudu prefetching cannot
be null");
+ this.keepAlivePeriodMs =
+ checkNotNull(keepAlivePeriodMs, "Kudu keep alive period ms
cannot be null");
+ this.replicaSelection =
+ checkNotNull(replicaSelection, "Kudu replica selection cannot
be null");
}
public String getMasters() {
@@ -50,26 +74,69 @@ public class KuduReaderConfig implements Serializable {
return rowLimit;
}
+ public long getSplitSizeBytes() {
+ return splitSizeBytes;
+ }
+
+ public int getBatchSizeBytes() {
+ return batchSizeBytes;
+ }
+
+ public long getScanRequestTimeout() {
+ return scanRequestTimeout;
+ }
+
+ public boolean isPrefetching() {
+ return prefetching;
+ }
+
+ public long getKeepAlivePeriodMs() {
+ return keepAlivePeriodMs;
+ }
+
+ public ReplicaSelection getReplicaSelection() {
+ return replicaSelection;
+ }
+
@Override
public String toString() {
return new ToStringBuilder(this)
.append("masters", masters)
.append("rowLimit", rowLimit)
+ .append("splitSizeBytes", splitSizeBytes)
+ .append("batchSizeBytes", batchSizeBytes)
+ .append("scanRequestTimeout", scanRequestTimeout)
+ .append("prefetching", prefetching)
+ .append("keepAlivePeriodMs", keepAlivePeriodMs)
+ .append("replicaSelection", replicaSelection)
.toString();
}
/** Builder for the {@link KuduReaderConfig}. */
public static class Builder {
- private static final int DEFAULT_ROW_LIMIT = 0;
+ // Reference from AbstractKuduScannerBuilder limit Long.MAX_VALUE
+ private static final int DEFAULT_ROW_LIMIT = Integer.MAX_VALUE;
private final String masters;
private final int rowLimit;
+ // Reference from KuduScanTokenBuilder splitSizeBytes
DEFAULT_SPLIT_SIZE_BYTES -1
+ private long splitSizeBytes = -1;
+ // Reference from BackupOptions DefaultScanBatchSize 1024 * 1024 * 20
+ private int batchSizeBytes = 1024 * 1024 * 20; // 20 MiB
+ // Reference from AsyncKuduClient DEFAULT_OPERATION_TIMEOUT_MS 30000
+ private long scanRequestTimeout = 30000;
+ // Reference from BackupOptions DefaultScanPrefetching false
+ private boolean prefetching = false;
+ // Reference from AsyncKuduClient DEFAULT_KEEP_ALIVE_PERIOD_MS 15000
+ private long keepAlivePeriodMs = 15000;
+ // Reference from BackupOptions DefaultScanLeaderOnly false
+ private ReplicaSelection replicaSelection =
ReplicaSelection.CLOSEST_REPLICA;
private Builder(String masters) {
this(masters, DEFAULT_ROW_LIMIT);
}
- private Builder(String masters, Integer rowLimit) {
+ private Builder(String masters, int rowLimit) {
this.masters = masters;
this.rowLimit = rowLimit;
}
@@ -82,8 +149,46 @@ public class KuduReaderConfig implements Serializable {
return new Builder(masters, rowLimit);
}
+ public Builder setSplitSizeBytes(long splitSizeBytes) {
+ this.splitSizeBytes = splitSizeBytes;
+ return this;
+ }
+
+ public Builder setBatchSizeBytes(int batchSizeBytes) {
+ this.batchSizeBytes = batchSizeBytes;
+ return this;
+ }
+
+ public Builder setScanRequestTimeout(long scanRequestTimeout) {
+ this.scanRequestTimeout = scanRequestTimeout;
+ return this;
+ }
+
+ public Builder setPrefetching(boolean prefetching) {
+ this.prefetching = prefetching;
+ return this;
+ }
+
+ public Builder setKeepAlivePeriodMs(long keepAlivePeriodMs) {
+ this.keepAlivePeriodMs = keepAlivePeriodMs;
+ return this;
+ }
+
+ public Builder setReplicaSelection(ReplicaSelection replicaSelection) {
+ this.replicaSelection = replicaSelection;
+ return this;
+ }
+
public KuduReaderConfig build() {
- return new KuduReaderConfig(masters, rowLimit);
+ return new KuduReaderConfig(
+ masters,
+ rowLimit,
+ splitSizeBytes,
+ batchSizeBytes,
+ scanRequestTimeout,
+ prefetching,
+ keepAlivePeriodMs,
+ replicaSelection);
}
}
}
diff --git
a/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/KuduSource.java
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/KuduSource.java
new file mode 100644
index 0000000..9e5d14b
--- /dev/null
+++
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/KuduSource.java
@@ -0,0 +1,133 @@
+/*
+ * 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 Flink {@link Source} for reading data from Apache Kudu. It supports both
bounded and continuous
+ * unbounded modes, allowing it to function as a snapshot reader or as a
CDC-like source that
+ * captures ongoing changes.
+ *
+ * <ul>
+ * <li>If {@code Boundedness.BOUNDED} is set, the source reads a snapshot of
the table at a given
+ * point in time and emits only records from this fixed dataset.
+ * <li>If {@code Boundedness.CONTINUOUS_UNBOUNDED} is set, the source
performs differential scans
+ * at configured intervals to continuously capture changes.
+ * </ul>
+ *
+ * <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 Boundedness} - Specifies whether the source behaves in bounded
or unbounded mode.
+ * <li>{@link Duration} - Defines the polling interval, i.e. the time
between consecutive scans.
+ * <li>{@link RowResultConverter} - Converts Kudu's {@code RowResult} into
the desired output type
+ * {@code OUT}.
+ * </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 Boundedness boundedness;
+ private final Duration discoveryPeriod;
+ private final RowResultConverter<OUT> rowResultConverter;
+
+ private final Configuration configuration;
+
+ KuduSource(
+ KuduReaderConfig readerConfig,
+ KuduTableInfo tableInfo,
+ Boundedness boundedness,
+ Duration discoveryPeriod,
+ RowResultConverter<OUT> rowResultConverter) {
+ this.tableInfo = tableInfo;
+ this.readerConfig = readerConfig;
+ this.boundedness = boundedness;
+ this.discoveryPeriod = discoveryPeriod;
+ this.rowResultConverter = rowResultConverter;
+ this.configuration = new Configuration();
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return boundedness;
+ }
+
+ @Override
+ public SplitEnumerator<KuduSourceSplit, KuduSourceEnumeratorState>
createEnumerator(
+ SplitEnumeratorContext<KuduSourceSplit> enumContext) {
+ return new KuduSourceEnumerator(
+ tableInfo, readerConfig, boundedness, discoveryPeriod,
enumContext);
+ }
+
+ @Override
+ public SplitEnumerator<KuduSourceSplit, KuduSourceEnumeratorState>
restoreEnumerator(
+ SplitEnumeratorContext<KuduSourceSplit> enumContext,
+ KuduSourceEnumeratorState checkpoint)
+ throws Exception {
+ return new KuduSourceEnumerator(
+ tableInfo, readerConfig, boundedness, discoveryPeriod,
enumContext, checkpoint);
+ }
+
+ @Override
+ public SimpleVersionedSerializer<KuduSourceSplit> getSplitSerializer() {
+ return new KuduSourceSplitSerializer();
+ }
+
+ @Override
+ public SimpleVersionedSerializer<KuduSourceEnumeratorState>
+ getEnumeratorCheckpointSerializer() {
+ return new KuduSourceEnumeratorStateSerializer();
+ }
+
+ @Override
+ public SourceReader<OUT, KuduSourceSplit> createReader(SourceReaderContext
readerContext)
+ throws Exception {
+ return new KuduSourceReader<>(
+ () -> new KuduSourceSplitReader(readerConfig),
+ configuration,
+ readerContext,
+ rowResultConverter);
+ }
+}
diff --git
a/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/KuduSourceBuilder.java
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/KuduSourceBuilder.java
new file mode 100644
index 0000000..0d1eeca
--- /dev/null
+++
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/KuduSourceBuilder.java
@@ -0,0 +1,80 @@
+/*
+ * 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.api.connector.source.Boundedness;
+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.checkNotNull;
+
+/**
+ * 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 Boundedness boundedness = Boundedness.BOUNDED;
+ private Duration discoveryPeriod = null;
+ private RowResultConverter<OUT> rowResultConverter;
+
+ 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> setBoundedness(Boundedness boundedness) {
+ this.boundedness = boundedness;
+ return this;
+ }
+
+ public KuduSourceBuilder<OUT> setDiscoveryPeriod(Duration discoveryPeriod)
{
+ this.discoveryPeriod = discoveryPeriod;
+ return this;
+ }
+
+ public KuduSource<OUT> build() {
+ checkNotNull(tableInfo, "Table info must be provided.");
+ checkNotNull(readerConfig, "Reader config must be provided.");
+ checkNotNull(rowResultConverter, "RowResultConverter must be
provided.");
+ if (boundedness == Boundedness.CONTINUOUS_UNBOUNDED) {
+ checkNotNull(
+ discoveryPeriod,
+ "Discovery period must be provided for
CONTINUOUS_UNBOUNDED mode.");
+ }
+
+ return new KuduSource<>(
+ readerConfig, tableInfo, boundedness, discoveryPeriod,
rowResultConverter);
+ }
+}
diff --git
a/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumerator.java
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumerator.java
new file mode 100644
index 0000000..dbd6edc
--- /dev/null
+++
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumerator.java
@@ -0,0 +1,260 @@
+/*
+ * 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.Boundedness;
+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.flink.connector.kudu.source.utils.KuduSourceUtils;
+import org.apache.flink.connector.kudu.source.utils.KuduSplitGenerator;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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 static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The Kudu source enumerator is responsible for discovering and assigning
splits.
+ *
+ * <ul>
+ * <li>If {@code Boundedness.BOUNDED} is set, the enumerator generates
splits corresponding to the
+ * current snapshot time and emits records only for this bounded set.
+ * <li>If {@code Boundedness.CONTINUOUS_UNBOUNDED} is set, the enumerator
follows a CDC-like
+ * approach as described below.
+ * </ul>
+ *
+ * <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 KuduSplitGenerator splitGenerator;
+ private final Boundedness boundedness;
+ private final Duration discoveryInterval;
+
+ private long lastEndTimestamp;
+ private final List<KuduSourceSplit> unassigned;
+ private final List<KuduSourceSplit> pending;
+
+ public KuduSourceEnumerator(
+ KuduTableInfo tableInfo,
+ KuduReaderConfig readerConfig,
+ Boundedness boundedness,
+ Duration discoveryInterval,
+ SplitEnumeratorContext<KuduSourceSplit> context) {
+ this(
+ tableInfo,
+ readerConfig,
+ boundedness,
+ discoveryInterval,
+ context,
+ KuduSourceEnumeratorState.empty());
+ }
+
+ public KuduSourceEnumerator(
+ KuduTableInfo tableInfo,
+ KuduReaderConfig readerConfig,
+ Boundedness boundedness,
+ Duration discoveryInterval,
+ SplitEnumeratorContext<KuduSourceSplit> context,
+ KuduSourceEnumeratorState enumState) {
+ this.boundedness = checkNotNull(boundedness);
+ this.discoveryInterval = discoveryInterval;
+ this.context = checkNotNull(context);
+ this.readersAwaitingSplit = new ArrayList<>();
+ this.unassigned = enumState.getUnassigned();
+ this.pending = enumState.getPending();
+ this.splitGenerator = new KuduSplitGenerator(readerConfig, tableInfo);
+ this.lastEndTimestamp = enumState.getLastEndTimestamp();
+
+ validateConfiguration();
+ }
+
+ private void validateConfiguration() {
+ if (boundedness == Boundedness.CONTINUOUS_UNBOUNDED &&
discoveryInterval == null) {
+ throw new IllegalArgumentException(
+ "discoveryInterval must be set for CONTINUOUS_UNBOUNDED
mode.");
+ }
+ }
+
+ @Override
+ public void start() {
+ if (boundedness == Boundedness.CONTINUOUS_UNBOUNDED) {
+ context.callAsync(
+ this::enumerateNewSplits, this::assignSplits, 0,
discoveryInterval.toMillis());
+ } else if (boundedness == Boundedness.BOUNDED) {
+ context.callAsync(this::enumerateNewSplits, this::assignSplits);
+ }
+ }
+
+ @Override
+ public void handleSplitRequest(int subtaskId, @Nullable String
requesterHostname) {
+ readersAwaitingSplit.add(subtaskId);
+ assignSplitsToReaders();
+ }
+
+ @Override
+ public void addSplitsBack(List<KuduSourceSplit> splits, int subtaskId) {
+ LOG.debug("Adding splits back: {}", splits);
+ pending.removeAll(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);
+ assignSplitsToReaders();
+ }
+ }
+
+ // 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() {
+ if (!shouldEnumerateNewSplits()) {
+ return null;
+ }
+ List<KuduSourceSplit> newSplits;
+
+ if (isFirstSplitGeneration()) {
+ lastEndTimestamp = KuduSourceUtils.getCurrentHybridTime();
+ newSplits =
splitGenerator.generateFullScanSplits(lastEndTimestamp);
+ } else {
+ long startHT = lastEndTimestamp;
+ long endHT = KuduSourceUtils.getCurrentHybridTime();
+ newSplits = splitGenerator.generateIncrementalSplits(startHT,
endHT);
+ lastEndTimestamp = endHT;
+ }
+
+ return newSplits;
+ }
+
+ private void assignSplits(List<KuduSourceSplit> splits, Throwable error) {
+ if (error != null) {
+ throw new FlinkRuntimeException(
+ "Failed to enumerate Kudu splits, shutting down job.",
error);
+ }
+
+ if (splits != null) {
+ 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 KuduSourceSplit nextSplit =
KuduSourceUtils.getNextSplit(unassigned);
+ if (nextSplit != null) {
+ context.assignSplit(nextSplit, awaitingSubtask);
+ awaitingSubtasks.remove();
+ pending.add(nextSplit);
+ } else {
+ if (boundedness == Boundedness.BOUNDED) {
+ awaitingSubtasks.remove();
+ context.signalNoMoreSplits(awaitingSubtask);
+ }
+ break;
+ }
+ }
+ }
+
+ private Boolean shouldEnumerateNewSplits() {
+ return pending.isEmpty() && unassigned.isEmpty();
+ }
+
+ private Boolean isFirstSplitGeneration() {
+ return lastEndTimestamp == -1L;
+ }
+}
diff --git
a/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumeratorState.java
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumeratorState.java
new file mode 100644
index 0000000..b784ae9
--- /dev/null
+++
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumeratorState.java
@@ -0,0 +1,57 @@
+/*
+ * 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.ArrayList;
+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(
+ long lastEndTimestamp,
+ List<KuduSourceSplit> unassigned,
+ List<KuduSourceSplit> pending) {
+ this.lastEndTimestamp = lastEndTimestamp;
+ this.unassigned = unassigned;
+ this.pending = pending;
+ }
+
+ public static KuduSourceEnumeratorState empty() {
+ return new KuduSourceEnumeratorState(-1L, new ArrayList<>(), new
ArrayList<>());
+ }
+
+ long getLastEndTimestamp() {
+ return lastEndTimestamp;
+ }
+
+ List<KuduSourceSplit> getUnassigned() {
+ return unassigned;
+ }
+
+ List<KuduSourceSplit> getPending() {
+ return pending;
+ }
+}
diff --git
a/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumeratorStateSerializer.java
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumeratorStateSerializer.java
new file mode 100644
index 0000000..a34ca40
--- /dev/null
+++
b/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 CURRENT_VERSION = 1;
+ private final KuduSourceSplitSerializer splitSerializer = new
KuduSourceSplitSerializer();
+
+ @Override
+ public int getVersion() {
+ return CURRENT_VERSION;
+ }
+
+ @Override
+ public byte[] serialize(KuduSourceEnumeratorState state) throws
IOException {
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(baos)) {
+
+ out.writeLong(state.getLastEndTimestamp());
+ serializeSplits(out, state.getUnassigned());
+ serializeSplits(out, state.getPending());
+
+ return baos.toByteArray();
+ }
+ }
+
+ @Override
+ public KuduSourceEnumeratorState deserialize(int version, byte[]
serialized)
+ throws IOException {
+ if (version != CURRENT_VERSION) {
+ throw new IllegalArgumentException("Unsupported version: " +
version);
+ }
+
+ try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
+ DataInputStream in = new DataInputStream(bais)) {
+
+ long lastEndTimestamp = in.readLong();
+ List<KuduSourceSplit> unassigned = deserializeSplits(in);
+ List<KuduSourceSplit> pending = deserializeSplits(in);
+
+ return new KuduSourceEnumeratorState(lastEndTimestamp, unassigned,
pending);
+ }
+ }
+
+ private void serializeSplits(DataOutputStream out, List<KuduSourceSplit>
splits)
+ throws IOException {
+ out.writeInt(splits.size());
+ for (KuduSourceSplit split : splits) {
+ byte[] splitBytes = splitSerializer.serialize(split);
+ out.writeInt(splitBytes.length);
+ out.write(splitBytes);
+ }
+ }
+
+ private List<KuduSourceSplit> deserializeSplits(DataInputStream in) throws
IOException {
+ int size = in.readInt();
+ List<KuduSourceSplit> splits = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ int length = in.readInt();
+ byte[] splitBytes = new byte[length];
+ in.readFully(splitBytes);
+ splits.add(splitSerializer.deserialize(CURRENT_VERSION,
splitBytes));
+ }
+ return splits;
+ }
+}
diff --git
a/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/reader/KuduRecordEmitter.java
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/reader/KuduRecordEmitter.java
new file mode 100644
index 0000000..f4f5d54
--- /dev/null
+++
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/reader/KuduRecordEmitter.java
@@ -0,0 +1,43 @@
+/*
+ * 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.reader;
+
+import org.apache.flink.api.connector.source.SourceOutput;
+import org.apache.flink.connector.base.source.reader.RecordEmitter;
+import org.apache.flink.connector.kudu.connector.converter.RowResultConverter;
+import org.apache.flink.connector.kudu.source.split.KuduSourceSplit;
+
+import org.apache.kudu.client.RowResult;
+
+/**
+ * The Kudu record emitter.
+ *
+ * @param <T> The type of the record.
+ */
+public class KuduRecordEmitter<T> implements RecordEmitter<RowResult, T,
KuduSourceSplit> {
+ private final RowResultConverter<T> rowResultConverter;
+
+ public KuduRecordEmitter(RowResultConverter<T> rowResultConverter) {
+ this.rowResultConverter = rowResultConverter;
+ }
+
+ @Override
+ public void emitRecord(RowResult element, SourceOutput<T> output,
KuduSourceSplit splitState) {
+ output.collect(rowResultConverter.convert(element));
+ }
+}
diff --git
a/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/reader/KuduSourceReader.java
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/reader/KuduSourceReader.java
new file mode 100644
index 0000000..9fd87d5
--- /dev/null
+++
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/reader/KuduSourceReader.java
@@ -0,0 +1,73 @@
+/*
+ * 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.reader;
+
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.configuration.Configuration;
+import
org.apache.flink.connector.base.source.reader.SingleThreadMultiplexSourceReaderBase;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
+import org.apache.flink.connector.kudu.connector.converter.RowResultConverter;
+import org.apache.flink.connector.kudu.source.split.KuduSourceSplit;
+import org.apache.flink.connector.kudu.source.split.SplitFinishedEvent;
+
+import org.apache.kudu.client.RowResult;
+
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.function.Supplier;
+
+/**
+ * The Kudu source reader.
+ *
+ * @param <OUT> the type of records read from the source.
+ */
+public class KuduSourceReader<OUT>
+ extends SingleThreadMultiplexSourceReaderBase<
+ RowResult, OUT, KuduSourceSplit, KuduSourceSplit> {
+
+ public KuduSourceReader(
+ Supplier<SplitReader<RowResult, KuduSourceSplit>>
splitReaderSupplier,
+ Configuration config,
+ SourceReaderContext context,
+ RowResultConverter<OUT> rowResultConverter) {
+ super(splitReaderSupplier, new
KuduRecordEmitter<>(rowResultConverter), config, context);
+ }
+
+ @Override
+ protected void onSplitFinished(Map<String, KuduSourceSplit>
finishedSplits) {
+ context.sendSourceEventToCoordinator(
+ new SplitFinishedEvent(new
ArrayList<>(finishedSplits.values())));
+ }
+
+ @Override
+ public void start() {
+ if (getNumberOfCurrentlyAssignedSplits() == 0) {
+ context.sendSplitRequest();
+ }
+ }
+
+ @Override
+ protected KuduSourceSplit initializedState(KuduSourceSplit split) {
+ return split;
+ }
+
+ @Override
+ protected KuduSourceSplit toSplitType(String splitId, KuduSourceSplit
splitState) {
+ return splitState;
+ }
+}
diff --git
a/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/reader/KuduSourceSplitReader.java
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/reader/KuduSourceSplitReader.java
new file mode 100644
index 0000000..f7bf5d5
--- /dev/null
+++
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/reader/KuduSourceSplitReader.java
@@ -0,0 +1,105 @@
+/*
+ * 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.reader;
+
+import org.apache.flink.connector.base.source.reader.RecordsBySplits;
+import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
+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.utils.KuduSourceUtils;
+
+import org.apache.kudu.client.KuduClient;
+import org.apache.kudu.client.KuduScanToken;
+import org.apache.kudu.client.KuduScanner;
+import org.apache.kudu.client.RowResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/** The Kudu source reader that reads data for corresponding splits. */
+public class KuduSourceSplitReader implements SplitReader<RowResult,
KuduSourceSplit> {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(KuduSourceSplitReader.class);
+ private final KuduClient kuduClient;
+ private final List<KuduSourceSplit> splits;
+ private final AtomicBoolean wakeUpFlag = new AtomicBoolean(false);
+
+ public KuduSourceSplitReader(KuduReaderConfig readerConfig) {
+ this.kuduClient = new
KuduClient.KuduClientBuilder(readerConfig.getMasters()).build();
+ this.splits = new ArrayList<>();
+ }
+
+ @Override
+ public RecordsWithSplitIds<RowResult> fetch() throws IOException {
+ wakeUpFlag.compareAndSet(true, false);
+
+ final KuduSourceSplit currentSplit =
KuduSourceUtils.getNextSplit(splits);
+ if (currentSplit == null) {
+ return new RecordsBySplits.Builder<RowResult>().build();
+ }
+
+ byte[] serializedToken = currentSplit.getSerializedScanToken();
+ KuduScanner scanner =
KuduScanToken.deserializeIntoScanner(serializedToken, kuduClient);
+ RecordsBySplits.Builder<RowResult> builder = new
RecordsBySplits.Builder<>();
+
+ try {
+ while (scanner.hasMoreRows()) {
+ for (RowResult row : scanner.nextRows()) {
+ if (wakeUpFlag.get()) {
+ LOG.debug("Wakeup signal received inside row
iteration, stopping fetch.");
+ scanner.close(); // Close the scanner
+ splits.add(currentSplit); // Put the split back
+ return new RecordsBySplits.Builder<RowResult>()
+ .build(); // Return empty result
+ }
+ builder.add(currentSplit.splitId(), row);
+ }
+ }
+ builder.addFinishedSplit(
+ currentSplit.splitId()); // Mark split as completed only
after the loop
+
+ } finally {
+ scanner.close(); // Ensure scanner is always closed
+ }
+
+ return builder.build();
+ }
+
+ @Override
+ public void handleSplitsChanges(SplitsChange<KuduSourceSplit>
splitsChanges) {
+ LOG.debug("Handling split change {}", splitsChanges);
+ splits.addAll(splitsChanges.splits());
+ }
+
+ @Override
+ public void wakeUp() {
+ LOG.debug("Wakeup called, setting flag.");
+ wakeUpFlag.set(true);
+ }
+
+ @Override
+ public void close() throws Exception {
+ kuduClient.close();
+ }
+}
diff --git
a/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/split/KuduSourceSplit.java
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/split/KuduSourceSplit.java
new file mode 100644
index 0000000..a57d398
--- /dev/null
+++
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/split/KuduSourceSplit.java
@@ -0,0 +1,70 @@
+/*
+ * 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 org.apache.commons.codec.digest.DigestUtils;
+
+import java.io.Serializable;
+import java.util.Arrays;
+
+/** 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.
+ */
+ private final byte[] serializedScanToken;
+
+ public KuduSourceSplit(byte[] serializedScanToken) {
+ this.serializedScanToken = serializedScanToken;
+ }
+
+ public byte[] getSerializedScanToken() {
+ return serializedScanToken;
+ }
+
+ @Override
+ public String splitId() {
+ return DigestUtils.sha256Hex(serializedScanToken);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ KuduSourceSplit that = (KuduSourceSplit) obj;
+ return Arrays.equals(serializedScanToken, that.serializedScanToken);
+ }
+
+ @Override
+ public int hashCode() {
+ return Arrays.hashCode(serializedScanToken);
+ }
+
+ @Override
+ public String toString() {
+ return "KuduSourceSplit{" + "splitId='" + splitId() + "'" + '}';
+ }
+}
diff --git
a/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/split/KuduSourceSplitSerializer.java
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/split/KuduSourceSplitSerializer.java
new file mode 100644
index 0000000..1dea4a6
--- /dev/null
+++
b/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 CURRENT_VERSION = 1; // Versioning for future
changes
+
+ @Override
+ public int getVersion() {
+ return CURRENT_VERSION;
+ }
+
+ @Override
+ public byte[] serialize(KuduSourceSplit obj) throws IOException {
+ if (obj == null || obj.getSerializedScanToken() == null) {
+ throw new IOException("KuduSourceSplit or serializedScanToken is
null.");
+ }
+
+ return obj.getSerializedScanToken(); // Directly return the byte array
+ }
+
+ @Override
+ public KuduSourceSplit deserialize(int version, byte[] serialized) throws
IOException {
+ if (version != CURRENT_VERSION) {
+ throw new IllegalArgumentException("Unsupported version: " +
version);
+ }
+ if (serialized == null || serialized.length == 0) {
+ throw new IOException("Serialized data is empty or null.");
+ }
+
+ return new KuduSourceSplit(serialized); // Recreate the split with the
byte array
+ }
+}
diff --git
a/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/split/SplitFinishedEvent.java
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/split/SplitFinishedEvent.java
new file mode 100644
index 0000000..6bb019b
--- /dev/null
+++
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/split/SplitFinishedEvent.java
@@ -0,0 +1,42 @@
+/*
+ * 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.SourceEvent;
+
+import java.util.List;
+
+/**
+ * A source event use to signal from {@link
+ * org.apache.flink.connector.kudu.source.reader.KuduSourceReader} to {@link
+ * org.apache.flink.connector.kudu.source.enumerator.KuduSourceEnumerator} the
finished splits. This
+ * allows us to differentiate between enumerated but unassigned and pending
splits in the
+ * enumerator.
+ */
+public class SplitFinishedEvent implements SourceEvent {
+
+ private final List<KuduSourceSplit> finishedSplits;
+
+ public SplitFinishedEvent(List<KuduSourceSplit> finishedSplits) {
+ this.finishedSplits = finishedSplits;
+ }
+
+ public List<KuduSourceSplit> getFinishedSplits() {
+ return finishedSplits;
+ }
+}
diff --git
a/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/utils/KuduSourceUtils.java
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/utils/KuduSourceUtils.java
new file mode 100644
index 0000000..2daa50d
--- /dev/null
+++
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/utils/KuduSourceUtils.java
@@ -0,0 +1,49 @@
+/*
+ * 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.utils;
+
+import org.apache.flink.connector.kudu.source.split.KuduSourceSplit;
+
+import org.apache.kudu.util.HybridTimeUtil;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.TimeUnit;
+
+/** Utility class for retrieving the next available Kudu source split. */
+public class KuduSourceUtils {
+
+ private KuduSourceUtils() {}
+
+ /** Retrieves and removes the next available split from the given
collection. */
+ public static KuduSourceSplit getNextSplit(Collection<KuduSourceSplit>
splits) {
+ if (splits == null || splits.isEmpty()) {
+ return null;
+ }
+ Iterator<KuduSourceSplit> iterator = splits.iterator();
+ KuduSourceSplit next = iterator.next();
+ iterator.remove();
+ return next;
+ }
+
+ public static long getCurrentHybridTime() {
+ return HybridTimeUtil.clockTimestampToHTTimestamp(
+ System.currentTimeMillis(), TimeUnit.MILLISECONDS)
+ + 1;
+ }
+}
diff --git
a/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/utils/KuduSplitGenerator.java
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/utils/KuduSplitGenerator.java
new file mode 100644
index 0000000..a4baac4
--- /dev/null
+++
b/flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/utils/KuduSplitGenerator.java
@@ -0,0 +1,122 @@
+/*
+ * 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.utils;
+
+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.kudu.client.AsyncKuduScanner;
+import org.apache.kudu.client.KuduClient;
+import org.apache.kudu.client.KuduException;
+import org.apache.kudu.client.KuduScanToken;
+import org.apache.kudu.client.KuduTable;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * The class responsible for producing scan tokens for given timestamps and
returning them in the
+ * form of {@link KuduSourceSplit}.
+ */
+public class KuduSplitGenerator implements AutoCloseable {
+ private final KuduTableInfo tableInfo;
+ private final KuduClient kuduClient;
+ private final KuduReaderConfig readerConfig;
+
+ public KuduSplitGenerator(KuduReaderConfig readerConfig, KuduTableInfo
tableInfo) {
+ this.tableInfo = tableInfo;
+ this.readerConfig = readerConfig;
+ this.kuduClient = new
KuduClient.KuduClientBuilder(readerConfig.getMasters()).build();
+ }
+
+ public List<KuduSourceSplit> generateFullScanSplits(long
snapshotTimestamp) {
+ if (snapshotTimestamp <= 0) {
+ throw new IllegalArgumentException(
+ "Snapshot timestamp must be greater than 0, but was: " +
snapshotTimestamp);
+ }
+ try {
+ List<KuduScanToken> tokens =
+ obtainScanTokenBuilder(tableInfo.getName())
+ .snapshotTimestampRaw(snapshotTimestamp)
+
.readMode(AsyncKuduScanner.ReadMode.READ_AT_SNAPSHOT)
+ .build();
+ return serializeTokens(tokens);
+ } catch (Exception e) {
+ throw new RuntimeException("Error during full snapshot scan: " +
e.getMessage(), e);
+ }
+ }
+
+ public List<KuduSourceSplit> generateIncrementalSplits(long startHT, long
endHT) {
+ if (startHT <= 0 || endHT <= 0) {
+ throw new IllegalArgumentException(
+ "Start and end timestamps must be greater than 0. Given
startHT: "
+ + startHT
+ + ", endHT: "
+ + endHT);
+ }
+
+ if (startHT >= endHT) {
+ throw new IllegalArgumentException(
+ "Start timestamp must be less than end timestamp. Given
startHT: "
+ + startHT
+ + ", endHT: "
+ + endHT);
+ }
+
+ try {
+ List<KuduScanToken> tokens =
+
obtainScanTokenBuilder(tableInfo.getName()).diffScan(startHT, endHT).build();
+ return serializeTokens(tokens);
+ } catch (Exception e) {
+ throw new RuntimeException("Error during incremental diff scan: "
+ e.getMessage(), e);
+ }
+ }
+
+ private List<KuduSourceSplit> serializeTokens(List<KuduScanToken> tokens) {
+ try {
+ List<KuduSourceSplit> splits = new ArrayList<>();
+ for (KuduScanToken token : tokens) {
+ splits.add(new KuduSourceSplit(token.serialize()));
+ }
+ return splits;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Error during source split serialization: " +
e.getMessage(), e);
+ }
+ }
+
+ private KuduScanToken.KuduScanTokenBuilder obtainScanTokenBuilder(String
tableName)
+ throws KuduException {
+ KuduTable table = kuduClient.openTable(tableName);
+ return kuduClient
+ .newScanTokenBuilder(table)
+ .limit(readerConfig.getRowLimit())
+ .setSplitSizeBytes(readerConfig.getSplitSizeBytes())
+ .batchSizeBytes(readerConfig.getBatchSizeBytes())
+ .scanRequestTimeout(readerConfig.getScanRequestTimeout())
+ .prefetching(readerConfig.isPrefetching())
+ .keepAlivePeriodMs(readerConfig.getKeepAlivePeriodMs())
+ .replicaSelection(readerConfig.getReplicaSelection());
+ }
+
+ @Override
+ public void close() throws Exception {
+ kuduClient.close();
+ }
+}
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/connector/KuduTestBase.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/connector/KuduTestBase.java
index 4ab1761..2b00efb 100644
---
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/connector/KuduTestBase.java
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/connector/KuduTestBase.java
@@ -93,7 +93,9 @@ public class KuduTestBase {
new GenericContainer<>(DOCKER_IMAGE)
.withExposedPorts(KUDU_MASTER_PORT, 8051)
.withCommand("master")
- .withEnv("MASTER_ARGS", "--unlock_unsafe_flags")
+ .withEnv(
+ "MASTER_ARGS",
+ "--unlock_unsafe_flags
--time_source=system_unsync --use_hybrid_clock=true")
.withNetwork(network)
.withNetworkAliases("kudu-master");
master.start();
@@ -111,7 +113,8 @@ public class KuduTestBase {
.withEnv(
"TSERVER_ARGS",
"--fs_wal_dir=/var/lib/kudu/tserver
--logtostderr --unlock_unsafe_flags"
- + " --use_hybrid_clock=false
--rpc_advertised_addresses="
+ + " --time_source=system_unsync
--use_hybrid_clock=true"
+ + " --rpc_advertised_addresses="
+ instanceName)
.withNetwork(network)
.withNetworkAliases(instanceName)
@@ -277,7 +280,7 @@ public class KuduTestBase {
return masterAddress.toString();
}
- public KuduClient getClient() {
+ public static KuduClient getClient() {
return kuduClient;
}
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/KuduSourceITCase.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/KuduSourceITCase.java
new file mode 100644
index 0000000..7b0c751
--- /dev/null
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/KuduSourceITCase.java
@@ -0,0 +1,132 @@
+/*
+ * 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.api.common.JobID;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.client.program.ClusterClient;
+import
org.apache.flink.connector.kudu.connector.converter.RowResultRowConverter;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.types.Row;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.time.Duration;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.function.Supplier;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for using Kudu Source. */
+public class KuduSourceITCase extends KuduSourceTestBase {
+ @RegisterExtension
+ private static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(2)
+ .build());
+
+ public static StreamExecutionEnvironment env;
+
+ private static Queue<Row> collectedRecords;
+
+ @BeforeEach
+ public void init() throws Exception {
+ super.init();
+ env = StreamExecutionEnvironment.getExecutionEnvironment();
+ collectedRecords = new ConcurrentLinkedDeque<>();
+ }
+
+ @Test
+ public void testRecordsFromSourceUnbounded(@InjectClusterClient
ClusterClient<?> client)
+ throws Exception {
+
+ KuduSource<Row> kuduSource =
+ new KuduSourceBuilder<Row>()
+ .setReaderConfig(getReaderConfig())
+ .setTableInfo(getTableInfo())
+ .setRowResultConverter(new RowResultRowConverter())
+ .setBoundedness(Boundedness.CONTINUOUS_UNBOUNDED)
+ .setDiscoveryPeriod(Duration.ofSeconds(1))
+ .build();
+
+ env.fromSource(kuduSource, WatermarkStrategy.noWatermarks(),
"KuduSource")
+ .returns(TypeInformation.of(Row.class))
+ .addSink(new TestingSinkFunction());
+
+ JobID jobID = env.executeAsync().getJobID();
+ waitExpectation(() -> collectedRecords.size() == 10);
+ assertThat(collectedRecords.size()).isEqualTo(10);
+ insertRows(10);
+ waitExpectation(() -> collectedRecords.size() == 20);
+ assertThat(collectedRecords.size()).isEqualTo(20);
+ client.cancel(jobID);
+ }
+
+ @Test
+ public void testRecordsFromSourceBounded() throws Exception {
+
+ KuduSource<Row> kuduSource =
+ new KuduSourceBuilder<Row>()
+ .setReaderConfig(getReaderConfig())
+ .setTableInfo(getTableInfo())
+ .setRowResultConverter(new RowResultRowConverter())
+ .build();
+
+ env.fromSource(kuduSource, WatermarkStrategy.noWatermarks(),
"KuduSource")
+ .returns(TypeInformation.of(Row.class))
+ .addSink(new TestingSinkFunction());
+
+ env.execute();
+ assertThat(collectedRecords.size()).isEqualTo(10);
+ }
+
+ private static void waitExpectation(Supplier<Boolean> condition) throws
Exception {
+ CompletableFuture<Void> future =
+ CompletableFuture.runAsync(
+ () -> {
+ while (!condition.get()) {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ });
+ future.get();
+ }
+
+ /** A sink function to collect the records. */
+ static class TestingSinkFunction implements SinkFunction<Row> {
+
+ @Override
+ public void invoke(Row value, Context context) throws Exception {
+ collectedRecords.add(value);
+ }
+ }
+}
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/KuduSourceTest.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/KuduSourceTest.java
new file mode 100644
index 0000000..aef9143
--- /dev/null
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/KuduSourceTest.java
@@ -0,0 +1,81 @@
+/*
+ * 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.api.connector.source.Boundedness;
+import org.apache.flink.connector.kudu.connector.KuduTableInfo;
+import
org.apache.flink.connector.kudu.connector.converter.RowResultRowConverter;
+import org.apache.flink.connector.kudu.connector.reader.KuduReaderConfig;
+import org.apache.flink.types.Row;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link KuduSource}. */
+public class KuduSourceTest {
+ @Test
+ void testNonExistentReaderConfig() {
+ KuduSourceBuilder<Row> builder =
+ new KuduSourceBuilder<Row>()
+ .setTableInfo(KuduTableInfo.forTable("table"))
+ .setRowResultConverter(new RowResultRowConverter());
+
+ assertThatThrownBy(builder::build)
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("Reader config");
+ }
+
+ @Test
+ void testNonExistentTableInfo() {
+ KuduSourceBuilder<Row> builder =
+ new KuduSourceBuilder<Row>()
+
.setReaderConfig(KuduReaderConfig.Builder.setMasters("masters").build())
+ .setRowResultConverter(new RowResultRowConverter());
+
+ assertThatThrownBy(builder::build)
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("Table info");
+ }
+
+ @Test
+ void testNonExistentRowResultConverter() {
+ KuduSourceBuilder<Row> builder =
+ new KuduSourceBuilder<Row>()
+
.setReaderConfig(KuduReaderConfig.Builder.setMasters("masters").build())
+ .setTableInfo(KuduTableInfo.forTable("table"));
+
+ assertThatThrownBy(builder::build)
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("RowResultConverter");
+ }
+
+ @Test
+ void testNonExistentDiscoveryIntervalContinuousMode() {
+ KuduSourceBuilder<Row> builder =
+ new KuduSourceBuilder<Row>()
+
.setReaderConfig(KuduReaderConfig.Builder.setMasters("masters").build())
+ .setTableInfo(KuduTableInfo.forTable("table"))
+ .setBoundedness(Boundedness.CONTINUOUS_UNBOUNDED)
+ .setRowResultConverter(new RowResultRowConverter());
+
+ assertThatThrownBy(builder::build)
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("Discovery period");
+ }
+}
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/KuduSourceTestBase.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/KuduSourceTestBase.java
new file mode 100644
index 0000000..4eba650
--- /dev/null
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/KuduSourceTestBase.java
@@ -0,0 +1,126 @@
+/*
+ * 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.KuduTestBase;
+import org.apache.flink.connector.kudu.connector.reader.KuduReaderConfig;
+
+import org.apache.kudu.ColumnSchema;
+import org.apache.kudu.Schema;
+import org.apache.kudu.Type;
+import org.apache.kudu.client.CreateTableOptions;
+import org.apache.kudu.client.Insert;
+import org.apache.kudu.client.KuduClient;
+import org.apache.kudu.client.KuduException;
+import org.apache.kudu.client.KuduSession;
+import org.apache.kudu.client.KuduTable;
+import org.apache.kudu.client.PartialRow;
+import org.apache.kudu.client.SessionConfiguration;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** A test base class for {@link KuduSource} related tests. */
+public class KuduSourceTestBase extends KuduTestBase {
+ private static KuduReaderConfig readerConfig;
+ private static KuduTableInfo tableInfo;
+ private static final String tableName = "test_table";
+ public static int lastInsertedKey = 0;
+
+ @BeforeEach
+ public void init() throws Exception {
+ readerConfig =
KuduReaderConfig.Builder.setMasters(getMasterAddress()).build();
+ tableInfo = KuduTableInfo.forTable(getTableName());
+
+ createExampleTable();
+ insertRows(10);
+ }
+
+ @AfterEach
+ public void tearDown() throws Exception {
+ deleteTable();
+ lastInsertedKey = 0;
+ }
+
+ public static KuduReaderConfig getReaderConfig() {
+ return readerConfig;
+ }
+
+ public static KuduTableInfo getTableInfo() {
+ return tableInfo;
+ }
+
+ public static String getTableName() {
+ return tableName;
+ }
+
+ public static int getTestRowsCount() {
+ return lastInsertedKey;
+ }
+
+ public static void createExampleTable() throws KuduException {
+ KuduClient client = getClient();
+ // Set up a simple schema.
+ List<ColumnSchema> columns = new ArrayList<>(2);
+ columns.add(new ColumnSchema.ColumnSchemaBuilder("key",
Type.INT32).key(true).build());
+ columns.add(
+ new ColumnSchema.ColumnSchemaBuilder("value",
Type.STRING).nullable(true).build());
+ Schema schema = new Schema(columns);
+
+ CreateTableOptions cto = new CreateTableOptions();
+ List<String> hashKeys = new ArrayList<>(1);
+ hashKeys.add("key");
+ int numBuckets = 2;
+ cto.addHashPartitions(hashKeys, numBuckets);
+
+ client.createTable(tableName, schema, cto);
+ }
+
+ public static void deleteTable() throws KuduException {
+ if (getClient().tableExists(tableName)) {
+ getClient().deleteTable(tableName);
+ }
+ }
+
+ public static void insertRows(int numRows) throws KuduException {
+ KuduTable table = getClient().openTable(tableName);
+ KuduSession session = getClient().newSession();
+
session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND);
+ for (int i = 0; i < numRows; i++) {
+ Insert insert = table.newInsert();
+ PartialRow row = insert.getRow();
+ int currentKey = lastInsertedKey++; // Increment key to ensure
uniqueness
+ row.addInt("key", currentKey);
+ // Make even-keyed row have a null 'value'.
+ if (i % 2 == 0) {
+ row.setNull("value");
+ } else {
+ row.addString("value", "value " + i);
+ }
+ session.apply(insert);
+ }
+
+ session.close();
+ if (session.countPendingErrors() != 0) {
+ throw new RuntimeException("Error inserting rows to Kudu");
+ }
+ }
+}
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumeratorStateSerializerTest.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumeratorStateSerializerTest.java
new file mode 100644
index 0000000..1c11c16
--- /dev/null
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumeratorStateSerializerTest.java
@@ -0,0 +1,85 @@
+/*
+ * 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.junit.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link KuduSourceEnumeratorStateSerializer}. */
+public class KuduSourceEnumeratorStateSerializerTest {
+ private final KuduSourceEnumeratorStateSerializer serializer =
+ new KuduSourceEnumeratorStateSerializer();
+
+ @Test
+ public void testSerializeDeserialize() throws IOException {
+ byte[] token1 = {1, 2, 3};
+ byte[] token2 = {4, 5, 6};
+ byte[] token3 = {7, 8, 9};
+ byte[] token4 = {10, 11, 12};
+
+ List<KuduSourceSplit> unassigned =
+ Arrays.asList(new KuduSourceSplit(token1), new
KuduSourceSplit(token2));
+ List<KuduSourceSplit> pending =
+ Arrays.asList(new KuduSourceSplit(token3), new
KuduSourceSplit(token4));
+ KuduSourceEnumeratorState state =
+ new KuduSourceEnumeratorState(12345L, unassigned, pending);
+
+ byte[] serialized = serializer.serialize(state);
+ KuduSourceEnumeratorState deserialized =
+ serializer.deserialize(serializer.getVersion(), serialized);
+
+
assertThat(deserialized.getLastEndTimestamp()).isEqualTo(state.getLastEndTimestamp());
+
assertThat(deserialized.getUnassigned()).hasSameSizeAs(state.getUnassigned());
+
assertThat(deserialized.getPending()).hasSameSizeAs(state.getPending());
+
+ for (int i = 0; i < unassigned.size(); i++) {
+ byte[] expected = unassigned.get(i).getSerializedScanToken();
+
assertThat(deserialized.getUnassigned().get(i).getSerializedScanToken())
+ .isEqualTo(expected);
+ }
+
+ for (int i = 0; i < pending.size(); i++) {
+ byte[] expected = pending.get(i).getSerializedScanToken();
+
assertThat(deserialized.getPending().get(i).getSerializedScanToken())
+ .isEqualTo(expected);
+ }
+ }
+
+ @Test
+ public void testSerializeDeserializeEmptyLists() throws IOException {
+ KuduSourceEnumeratorState state =
+ new KuduSourceEnumeratorState(
+ 67890L, Collections.emptyList(),
Collections.emptyList());
+
+ byte[] serialized = serializer.serialize(state);
+ KuduSourceEnumeratorState deserialized =
+ serializer.deserialize(serializer.getVersion(), serialized);
+
+
assertThat(state.getLastEndTimestamp()).isEqualTo(deserialized.getLastEndTimestamp());
+ assertThat(deserialized.getUnassigned()).isEmpty();
+ assertThat(deserialized.getPending()).isEmpty();
+ }
+}
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumeratorTest.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumeratorTest.java
new file mode 100644
index 0000000..814f159
--- /dev/null
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumeratorTest.java
@@ -0,0 +1,138 @@
+/*
+ * 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.Boundedness;
+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.testutils.source.reader.TestingSplitEnumeratorContext;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Unit tests for {@link KuduSourceEnumerator}.
+ *
+ * <p>Verifies that when a registered reader requests a split:
+ *
+ * <ul>
+ * <li>An unassigned split moves to the enumerator context's split
assignment.
+ * <li>The enumerator tracks the split by adding it to the pending list.
+ * </ul>
+ */
+public class KuduSourceEnumeratorTest {
+
+ private TestingSplitEnumeratorContext<KuduSourceSplit> context;
+ private KuduSourceSplit split;
+ private final int subtaskId = 1;
+ private final long checkpointId = 1L;
+ private final String requesterHostname = "host";
+ private final KuduTableInfo tableInfo = KuduTableInfo.forTable("table");
+ private final KuduReaderConfig readerConfig =
+ KuduReaderConfig.Builder.setMasters("master").build();
+
+ @BeforeEach
+ void setup() {
+ this.context = new TestingSplitEnumeratorContext<>(1);
+ }
+
+ private KuduSourceEnumerator createEnumerator(
+ SplitEnumeratorContext<KuduSourceSplit> context, Boundedness
boundedness) {
+ byte[] token = {1, 2, 3, 4, 5};
+ split = new KuduSourceSplit(token);
+
+ List<KuduSourceSplit> unassigned = new ArrayList<>();
+ unassigned.add(split);
+ List<KuduSourceSplit> pending = new ArrayList<>();
+
+ KuduSourceEnumeratorState state = new KuduSourceEnumeratorState(1L,
unassigned, pending);
+
+ return new KuduSourceEnumerator(
+ tableInfo, readerConfig, boundedness, Duration.ofSeconds(1),
context, state);
+ }
+
+ @ParameterizedTest
+ @EnumSource(Boundedness.class)
+ void testCheckpointNoSplitRequested(Boundedness boundedness) throws
Exception {
+ try (KuduSourceEnumerator enumerator = createEnumerator(context,
boundedness)) {
+ KuduSourceEnumeratorState state =
enumerator.snapshotState(checkpointId);
+ assertThat(state.getUnassigned().size()).isEqualTo(1);
+ assertThat(state.getPending().size()).isEqualTo(0);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to close KuduSourceEnumerator",
e);
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(Boundedness.class)
+ void testSplitRequestForRegisteredReader(Boundedness boundedness) throws
Exception {
+ try (KuduSourceEnumerator enumerator = createEnumerator(context,
boundedness)) {
+ context.registerReader(subtaskId, requesterHostname);
+ enumerator.addReader(subtaskId);
+ enumerator.handleSplitRequest(subtaskId, requesterHostname);
+
assertThat(enumerator.snapshotState(checkpointId).getUnassigned().size()).isEqualTo(0);
+
assertThat(enumerator.snapshotState(checkpointId).getPending().size()).isEqualTo(1);
+ assertThat(context.getSplitAssignments().size()).isEqualTo(1);
+
assertThat(context.getSplitAssignments().get(subtaskId).getAssignedSplits())
+ .contains(split);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to close KuduSourceEnumerator",
e);
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(Boundedness.class)
+ void testSplitRequestForNonRegisteredReader(Boundedness boundedness)
throws Exception {
+ try (KuduSourceEnumerator enumerator = createEnumerator(context,
boundedness)) {
+ enumerator.handleSplitRequest(subtaskId, requesterHostname);
+ assertThat(context.getSplitAssignments().size()).isEqualTo(0);
+
assertThat(enumerator.snapshotState(checkpointId).getUnassigned().size()).isEqualTo(1);
+
assertThat(enumerator.snapshotState(checkpointId).getUnassigned().get(0))
+ .isEqualTo(split);
+
assertThat(enumerator.snapshotState(checkpointId).getPending().size()).isEqualTo(0);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to close KuduSourceEnumerator",
e);
+ }
+ }
+
+ @Test
+ public void testCreateEnumeratorWithWrongConfig() {
+ assertThatThrownBy(
+ () ->
+ new KuduSourceEnumerator(
+ tableInfo,
+ readerConfig,
+ Boundedness.CONTINUOUS_UNBOUNDED,
+ null,
+ context,
+ KuduSourceEnumeratorState.empty()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("discoveryInterval must be set for
CONTINUOUS_UNBOUNDED mode.");
+ }
+}
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/enumerator/KuduSplitGeneratorTest.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/enumerator/KuduSplitGeneratorTest.java
new file mode 100644
index 0000000..8b79f66
--- /dev/null
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/enumerator/KuduSplitGeneratorTest.java
@@ -0,0 +1,154 @@
+/*
+ * 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.connector.KuduTableInfo;
+import org.apache.flink.connector.kudu.connector.reader.KuduReaderConfig;
+import org.apache.flink.connector.kudu.source.KuduSourceTestBase;
+import org.apache.flink.connector.kudu.source.split.KuduSourceSplit;
+import org.apache.flink.connector.kudu.source.utils.KuduSourceUtils;
+import org.apache.flink.connector.kudu.source.utils.KuduSplitGenerator;
+
+import org.apache.kudu.client.KuduScanToken;
+import org.apache.kudu.client.KuduScanner;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link KuduSplitGenerator}. */
+public class KuduSplitGeneratorTest extends KuduSourceTestBase {
+
+ private final long startHT;
+ private final long endHT;
+ private final long currentHT;
+
+ public KuduSplitGeneratorTest() throws Exception {
+ currentHT = KuduSourceUtils.getCurrentHybridTime();
+ startHT = KuduSourceUtils.getCurrentHybridTime();
+ Thread.sleep(1000);
+ endHT = KuduSourceUtils.getCurrentHybridTime();
+ }
+
+ @Test
+ public void testGenerateGoodFullScanSplits() {
+ try (KuduSplitGenerator generator =
+ new KuduSplitGenerator(getReaderConfig(), getTableInfo())) {
+ List<KuduSourceSplit> splits =
generator.generateFullScanSplits(currentHT);
+ assertThat(splits.size()).isGreaterThan(0);
+ // Check that all the splits can be properly deserialized into
Kudu scanners.
+ for (KuduSourceSplit split : splits) {
+ try {
+ KuduScanner scanner =
+ KuduScanToken.deserializeIntoScanner(
+ split.getSerializedScanToken(),
getClient());
+ } catch (Exception e) {
+ Assertions.fail("Deserialization failed: " +
e.getMessage());
+ }
+ }
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to close KuduSplitGenerator",
e);
+ }
+ }
+
+ @Test
+ public void testGenerateGoodIncrementalSplits() {
+ try (KuduSplitGenerator generator =
+ new KuduSplitGenerator(getReaderConfig(), getTableInfo())) {
+ List<KuduSourceSplit> splits =
generator.generateIncrementalSplits(startHT, endHT);
+ assertThat(splits.size()).isGreaterThan(0);
+ // Check that all the splits can be properly deserialized into
Kudu scanners.
+ for (KuduSourceSplit split : splits) {
+ try {
+ KuduScanner scanner =
+ KuduScanToken.deserializeIntoScanner(
+ split.getSerializedScanToken(),
getClient());
+ } catch (Exception e) {
+ Assertions.fail("Deserialization failed: " +
e.getMessage());
+ }
+ }
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to close KuduSplitGenerator",
e);
+ }
+ }
+
+ @Test
+ public void testInvalidReaderConfigFullScan() {
+ KuduReaderConfig invalidReaderConfig =
+ KuduReaderConfig.Builder.setMasters("invalidMasters").build();
+ try (KuduSplitGenerator generator =
+ new KuduSplitGenerator(invalidReaderConfig, getTableInfo())) {
+
+ assertThatThrownBy(() ->
generator.generateFullScanSplits(currentHT))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining(
+ "Error during full snapshot scan: Couldn't find a
valid master");
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to close KuduSplitGenerator",
e);
+ }
+ }
+
+ @Test
+ public void testInvalidTableInfoFullScan() {
+ KuduTableInfo invalidTableInfo =
KuduTableInfo.forTable("nonExistentTable");
+ try (KuduSplitGenerator generator =
+ new KuduSplitGenerator(getReaderConfig(), invalidTableInfo)) {
+
+ assertThatThrownBy(() ->
generator.generateFullScanSplits(currentHT))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining(
+ "Error during full snapshot scan: the table does
not exist");
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to close KuduSplitGenerator",
e);
+ }
+ }
+
+ @Test
+ public void testInvalidReaderConfigIncrementalScan() {
+ KuduReaderConfig invalidReaderConfig =
+ KuduReaderConfig.Builder.setMasters("invalidMasters").build();
+ try (KuduSplitGenerator generator =
+ new KuduSplitGenerator(invalidReaderConfig, getTableInfo())) {
+
+ assertThatThrownBy(() ->
generator.generateIncrementalSplits(startHT, endHT))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining(
+ "Error during incremental diff scan: Couldn't find
a valid master");
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to close KuduSplitGenerator",
e);
+ }
+ }
+
+ @Test
+ public void testInvalidTableInfoIncrementalScan() {
+ KuduTableInfo invalidTableInfo =
KuduTableInfo.forTable("nonExistentTable");
+ try (KuduSplitGenerator generator =
+ new KuduSplitGenerator(getReaderConfig(), invalidTableInfo)) {
+
+ assertThatThrownBy(() ->
generator.generateIncrementalSplits(startHT, endHT))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining(
+ "Error during incremental diff scan: the table
does not exist");
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to close KuduSplitGenerator",
e);
+ }
+ }
+}
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/reader/KuduSourceSplitReaderTest.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/reader/KuduSourceSplitReaderTest.java
new file mode 100644
index 0000000..12f3bc1
--- /dev/null
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/reader/KuduSourceSplitReaderTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.reader;
+
+import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
+import
org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition;
+import org.apache.flink.connector.kudu.source.KuduSourceTestBase;
+import org.apache.flink.connector.kudu.source.split.KuduSourceSplit;
+import org.apache.flink.connector.kudu.source.utils.KuduSourceUtils;
+import org.apache.flink.connector.kudu.source.utils.KuduSplitGenerator;
+
+import org.apache.kudu.client.RowResult;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link KuduSourceSplitReader}. */
+public class KuduSourceSplitReaderTest extends KuduSourceTestBase {
+
+ private List<KuduSourceSplit> generateSplits() {
+ List<KuduSourceSplit> splits;
+
+ try (KuduSplitGenerator generator =
+ new KuduSplitGenerator(getReaderConfig(), getTableInfo())) {
+ long now = KuduSourceUtils.getCurrentHybridTime();
+ splits = generator.generateFullScanSplits(now);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to close KuduSplitGenerator",
e);
+ }
+ return splits;
+ }
+
+ @Test
+ public void testBasicRecordFetching() {
+ int recordsFetched = 0;
+
+ try (KuduSourceSplitReader splitReader = new
KuduSourceSplitReader(getReaderConfig())) {
+ List<KuduSourceSplit> splits = generateSplits();
+ for (KuduSourceSplit split : splits) {
+ splitReader.handleSplitsChanges(
+ new SplitsAddition<>(new
ArrayList<>(Collections.singletonList(split))));
+ RecordsWithSplitIds<RowResult> fetchedRecordsWithSplitIds =
splitReader.fetch();
+ assertThat(fetchedRecordsWithSplitIds.nextSplit()).isNotNull();
+
+ List<RowResult> records = new ArrayList<>();
+ RowResult nextRecordFromSplit =
fetchedRecordsWithSplitIds.nextRecordFromSplit();
+ while (nextRecordFromSplit != null) {
+ records.add(nextRecordFromSplit);
+ nextRecordFromSplit =
fetchedRecordsWithSplitIds.nextRecordFromSplit();
+ }
+
+ assertThat(records.size()).isGreaterThan(0);
+ recordsFetched += records.size();
+ }
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to close
KuduSourceSplitReader", e);
+ }
+
+ assertThat(recordsFetched).isEqualTo(getTestRowsCount());
+ }
+}
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/split/KuduSourceSplitSerializerTest.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/split/KuduSourceSplitSerializerTest.java
new file mode 100644
index 0000000..b2ccd3d
--- /dev/null
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/source/split/KuduSourceSplitSerializerTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.junit.Test;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link KuduSourceSplitSerializer}. */
+public class KuduSourceSplitSerializerTest {
+
+ private final KuduSourceSplitSerializer splitSerializer = new
KuduSourceSplitSerializer();
+
+ @Test
+ public void testSplitSerializeDeserialize() throws IOException {
+ byte[] token = {1, 2, 3, 4, 5};
+ KuduSourceSplit split = new KuduSourceSplit(token);
+
+ byte[] serialized = splitSerializer.serialize(split);
+ KuduSourceSplit deserialized =
+ splitSerializer.deserialize(splitSerializer.getVersion(),
serialized);
+
+
assertThat(split.getSerializedScanToken()).isEqualTo(deserialized.getSerializedScanToken());
+ }
+
+ @Test
+ public void testSplitSerializationNullToken() {
+ assertThatThrownBy(() -> splitSerializer.serialize(null))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("KuduSourceSplit or serializedScanToken
is null.");
+ }
+
+ @Test
+ public void testSplitDeserializationEmptyData() {
+ assertThatThrownBy(
+ () ->
+ splitSerializer.deserialize(
+ splitSerializer.getVersion(), new
byte[] {}))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Serialized data is empty or null.");
+ }
+}
diff --git a/pom.xml b/pom.xml
index 3447e2e..6ad1355 100644
--- a/pom.xml
+++ b/pom.xml
@@ -111,6 +111,12 @@ under the License.
<artifactId>log4j-1.2-api</artifactId>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>commons-codec</groupId>
+ <artifactId>commons-codec</artifactId>
+ <version>1.17.1</version>
+ </dependency>
</dependencies>
<!-- This section defines the module versions that are used if nothing
else is specified. -->
@@ -122,6 +128,18 @@ under the License.
<version>${flink.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+
<artifactId>flink-connector-test-utils</artifactId>
+ <version>${flink.version}</version>
+ </dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
@@ -156,6 +174,20 @@ under the License.
<groupId>org.apache.flink</groupId>
<artifactId>flink-test-utils</artifactId>
<version>${flink.version}</version>
+ <exclusions>
+ <exclusion>
+
<groupId>org.apache.commons</groupId>
+
<artifactId>commons-compress</artifactId>
+ </exclusion>
+ <exclusion>
+
<groupId>org.xerial.snappy</groupId>
+
<artifactId>snappy-java</artifactId>
+ </exclusion>
+ <exclusion>
+
<groupId>org.apache.yetus</groupId>
+
<artifactId>audience-annotations</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<!-- Flink ArchUnit -->
