ferenc-csaky commented on code in PR #8:
URL:
https://github.com/apache/flink-connector-kudu/pull/8#discussion_r1975608209
##########
flink-connector-kudu/pom.xml:
##########
@@ -64,11 +69,22 @@ under the License.
<artifactId>flink-table-runtime</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils</artifactId>
Review Comment:
Lets add `test` to this too.
##########
flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumerator.java:
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.config.ContinuousBoundingSettings;
+import org.apache.flink.connector.kudu.source.split.KuduSourceSplit;
+import org.apache.flink.connector.kudu.source.split.SplitFinishedEvent;
+
+import org.apache.kudu.util.HybridTimeUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The Kudu source enumerator is responsible for periodically discovering and
assigning new splits
Review Comment:
Lets update the javadoc to reflect the `boundedness` related behavior
variations.
##########
flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/reader/KuduSourceReader.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.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 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<OUT, OUT,
KuduSourceSplit, KuduSourceSplit> {
+
+ public KuduSourceReader(
+ Supplier<SplitReader<OUT, KuduSourceSplit>> splitReaderSupplier,
+ Configuration config,
+ SourceReaderContext context,
+ RowResultConverter rowResultConverter) {
Review Comment:
This should be `RowResultConverter<OUT>`, which will be possible if the
`Supplier` is changed to the correct types.
##########
flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/reader/KuduSourceReader.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.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 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<OUT, OUT,
KuduSourceSplit, KuduSourceSplit> {
+
+ public KuduSourceReader(
+ Supplier<SplitReader<OUT, KuduSourceSplit>> splitReaderSupplier,
+ Configuration config,
+ SourceReaderContext context,
+ RowResultConverter rowResultConverter) {
+ super(splitReaderSupplier, new
KuduRecordEmitter<>(rowResultConverter), config, context);
+ }
+
+ @Override
+ protected void onSplitFinished(Map<String, KuduSourceSplit>
finishedSplits) {
Review Comment:
I'm wondering if a `context.sendSplitRequest()` should be called here as
well after the source event is sent.
##########
flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/testutils/KuduSourceITBase.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * 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.testutils;
+
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/**
+ * Need this in order for '@InjectClusterClient ClusterClient<?> client' to
work as test argument.
+ */
+public interface KuduSourceITBase {
Review Comment:
When a mini cluster is spinned up, that's more like an intergration test, so
I'd rather not mix that in the same class that contains unit tests. IMO this
should be transformed to a ` KuduSourceITCase` class, which contains the
relevant test cases and the mini cluster as a `static final` field.
##########
flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumerator.java:
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.config.ContinuousBoundingSettings;
+import org.apache.flink.connector.kudu.source.split.KuduSourceSplit;
+import org.apache.flink.connector.kudu.source.split.SplitFinishedEvent;
+
+import org.apache.kudu.util.HybridTimeUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The Kudu source enumerator is responsible for periodically discovering and
assigning new splits
+ * when possible.
+ *
+ * <p>To provide CDC-like functionality, the enumeration works as follows:
Initially, we perform a
+ * snapshot read of the table and mark the snapshot time as t0. From that
point onward, we perform
+ * differential scans in the time intervals t0 - t1, t1 - t2, and so on.
+ *
+ * <p>This approach means that new splits can only be enumerated once the
current time range is
+ * fully processed.
+ *
+ * <p>The process is controlled as follows:
+ *
+ * <ul>
+ * <li>Once a set of splits is enumerated for a time range, we track:
+ * <ul>
+ * <li><b>Unassigned splits</b>: Discovered but not yet assigned to
readers.
+ * <li><b>Pending splits</b>: Assigned but not yet fully processed.
+ * </ul>
+ * <li>A new set of splits is generated only when there are no remaining
unassigned or pending
+ * splits.
+ * </ul>
+ */
+public class KuduSourceEnumerator
+ implements SplitEnumerator<KuduSourceSplit, KuduSourceEnumeratorState>
{
+ private static final Logger LOG =
LoggerFactory.getLogger(KuduSourceEnumerator.class);
+
+ private final SplitEnumeratorContext<KuduSourceSplit> context;
+ private final List<Integer> readersAwaitingSplit;
+ private final KuduSplitGenerator splitGenerator;
+ private final ContinuousBoundingSettings continuousBoundingSettings;
+
+ private long lastEndTimestamp;
+ private final List<KuduSourceSplit> unassigned;
+ private final List<KuduSourceSplit> pending;
+
+ public KuduSourceEnumerator(
+ KuduTableInfo tableInfo,
+ KuduReaderConfig readerConfig,
+ ContinuousBoundingSettings continuousBoundingSettings,
+ SplitEnumeratorContext<KuduSourceSplit> context) {
+ this(
+ tableInfo,
+ readerConfig,
+ continuousBoundingSettings,
+ context,
+ KuduSourceEnumeratorState.empty());
+ }
+
+ public KuduSourceEnumerator(
+ KuduTableInfo tableInfo,
+ KuduReaderConfig readerConfig,
+ ContinuousBoundingSettings continuousBoundingSettings,
+ SplitEnumeratorContext<KuduSourceSplit> context,
+ KuduSourceEnumeratorState enumState) {
+ this.continuousBoundingSettings = continuousBoundingSettings;
+ 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();
+ }
+
+ @Override
+ public void start() {
+ if (continuousBoundingSettings.getBoundedness() ==
Boundedness.CONTINUOUS_UNBOUNDED
+ && Objects.nonNull(continuousBoundingSettings.getPeriod())) {
+ context.callAsync(
+ () -> enumerateNewSplits(() -> pending.isEmpty() &&
unassigned.isEmpty()),
+ this::assignSplits,
+ 0,
+ continuousBoundingSettings.getPeriod().toMillis());
+ } else {
+ List<KuduSourceSplit> splits = enumerateNewSplits(() -> true);
+ if (splits != null) {
+ unassigned.addAll(enumerateNewSplits(() -> true));
+ assignSplitsToReaders();
+ }
+ }
+ }
+
+ @Override
+ public void handleSplitRequest(int subtaskId, @Nullable String
requesterHostname) {
+ readersAwaitingSplit.add(subtaskId);
+ assignSplitsToReaders();
+ }
+
+ @Override
+ public void addSplitsBack(List<KuduSourceSplit> splits, int subtaskId) {
+ LOG.debug("File Source Enumerator adds splits back: {}", splits);
Review Comment:
nit: Message can be simplified to `Adding back splits: {}"` or something
like that.
##########
flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/reader/KuduSourceReader.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.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 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<OUT, OUT,
KuduSourceSplit, KuduSourceSplit> {
+
+ public KuduSourceReader(
+ Supplier<SplitReader<OUT, KuduSourceSplit>> splitReaderSupplier,
Review Comment:
This should be `Supplier<SplitReader<RowResult, KuduSourceSplit>>`.
##########
flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSplitGenerator.java:
##########
@@ -0,0 +1,109 @@
+/*
+ * 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.split.KuduSourceSplit;
+
+import org.apache.kudu.client.AsyncKuduScanner;
+import org.apache.kudu.client.KuduClient;
+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 {
Review Comment:
Since this is closeable, lets add `implements AutoCloseable`.
##########
flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/config/ContinuousBoundingSettings.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.config;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.connector.source.Boundedness;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.Objects;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Settings describing how to do continuous data discovery and enumeration for
the data source's
+ * continuous discovery and streaming mode.
+ */
+@PublicEvolving
+public class ContinuousBoundingSettings implements Serializable {
Review Comment:
I do not really see the necessity of this class (and especially its own
builder :)), I'd rather keep it simple and just add these 2 fields to
`KuduSource`, and `KuduSourceBuilder`. I'd set `boundedness` to `BOUNDED` and
`period` to `null` by default, and validate in `KuduSourcreBuilder#build()` if
`boundedness` is set to `CONTINUOUS_UNBOUNDED` then make sure `period` also has
a valid value.
IMO it's totally fine to pass these 2 variables on their own to the
`KuduSourceEnumerator` as well.
##########
flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumerator.java:
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.config.ContinuousBoundingSettings;
+import org.apache.flink.connector.kudu.source.split.KuduSourceSplit;
+import org.apache.flink.connector.kudu.source.split.SplitFinishedEvent;
+
+import org.apache.kudu.util.HybridTimeUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The Kudu source enumerator is responsible for periodically discovering and
assigning new splits
+ * when possible.
+ *
+ * <p>To provide CDC-like functionality, the enumeration works as follows:
Initially, we perform a
+ * snapshot read of the table and mark the snapshot time as t0. From that
point onward, we perform
+ * differential scans in the time intervals t0 - t1, t1 - t2, and so on.
+ *
+ * <p>This approach means that new splits can only be enumerated once the
current time range is
+ * fully processed.
+ *
+ * <p>The process is controlled as follows:
+ *
+ * <ul>
+ * <li>Once a set of splits is enumerated for a time range, we track:
+ * <ul>
+ * <li><b>Unassigned splits</b>: Discovered but not yet assigned to
readers.
+ * <li><b>Pending splits</b>: Assigned but not yet fully processed.
+ * </ul>
+ * <li>A new set of splits is generated only when there are no remaining
unassigned or pending
+ * splits.
+ * </ul>
+ */
+public class KuduSourceEnumerator
+ implements SplitEnumerator<KuduSourceSplit, KuduSourceEnumeratorState>
{
+ private static final Logger LOG =
LoggerFactory.getLogger(KuduSourceEnumerator.class);
+
+ private final SplitEnumeratorContext<KuduSourceSplit> context;
+ private final List<Integer> readersAwaitingSplit;
+ private final KuduSplitGenerator splitGenerator;
+ private final ContinuousBoundingSettings continuousBoundingSettings;
+
+ private long lastEndTimestamp;
+ private final List<KuduSourceSplit> unassigned;
+ private final List<KuduSourceSplit> pending;
+
+ public KuduSourceEnumerator(
+ KuduTableInfo tableInfo,
+ KuduReaderConfig readerConfig,
+ ContinuousBoundingSettings continuousBoundingSettings,
+ SplitEnumeratorContext<KuduSourceSplit> context) {
+ this(
+ tableInfo,
+ readerConfig,
+ continuousBoundingSettings,
+ context,
+ KuduSourceEnumeratorState.empty());
+ }
+
+ public KuduSourceEnumerator(
+ KuduTableInfo tableInfo,
+ KuduReaderConfig readerConfig,
+ ContinuousBoundingSettings continuousBoundingSettings,
+ SplitEnumeratorContext<KuduSourceSplit> context,
+ KuduSourceEnumeratorState enumState) {
+ this.continuousBoundingSettings = continuousBoundingSettings;
+ 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();
+ }
+
+ @Override
+ public void start() {
+ if (continuousBoundingSettings.getBoundedness() ==
Boundedness.CONTINUOUS_UNBOUNDED
+ && Objects.nonNull(continuousBoundingSettings.getPeriod())) {
+ context.callAsync(
+ () -> enumerateNewSplits(() -> pending.isEmpty() &&
unassigned.isEmpty()),
+ this::assignSplits,
+ 0,
+ continuousBoundingSettings.getPeriod().toMillis());
+ } else {
+ List<KuduSourceSplit> splits = enumerateNewSplits(() -> true);
+ if (splits != null) {
+ unassigned.addAll(enumerateNewSplits(() -> true));
+ assignSplitsToReaders();
+ }
+ }
+ }
+
+ @Override
+ public void handleSplitRequest(int subtaskId, @Nullable String
requesterHostname) {
+ readersAwaitingSplit.add(subtaskId);
+ assignSplitsToReaders();
+ }
+
+ @Override
+ public void addSplitsBack(List<KuduSourceSplit> splits, int subtaskId) {
+ LOG.debug("File Source Enumerator adds splits back: {}", splits);
+ unassigned.addAll(splits);
+ if (context.registeredReaders().containsKey(subtaskId)) {
+ readersAwaitingSplit.add(subtaskId);
+ }
+ assignSplitsToReaders();
+ }
+
+ @Override
+ public void addReader(int subtaskId) {
+ // The source is purely lazy-pull-based, nothing to do upon
registration
+ }
+
+ @Override
+ public KuduSourceEnumeratorState snapshotState(long checkpointId) throws
Exception {
+ return new KuduSourceEnumeratorState(lastEndTimestamp, unassigned,
pending);
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ splitGenerator.close();
+ } catch (Exception e) {
+ throw new IOException("Error closing split generator", e);
+ }
+ }
+
+ @Override
+ public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
+ if (sourceEvent instanceof SplitFinishedEvent) {
+ SplitFinishedEvent splitFinishedEvent = (SplitFinishedEvent)
sourceEvent;
+ LOG.debug(
+ "Received SplitFinishedEvent from subtask {} for splits:
{}",
+ subtaskId,
+ splitFinishedEvent.getFinishedSplits());
+ pending.removeAll(splitFinishedEvent.getFinishedSplits());
+ readersAwaitingSplit.add(subtaskId);
+ }
+ }
+
+ // This function is invoked repeatedly according to this.period if there
are no outstanding
+ // splits.
+ // Outstanding meaning that there are no pending splits, and no enumerated
but not assigned
+ // splits for the
+ // current period.
+ private List<KuduSourceSplit> enumerateNewSplits(Supplier<Boolean>
shouldGenerate) {
+ if (!shouldGenerate.get()) {
+ return null;
+ }
+ List<KuduSourceSplit> newSplits;
+
+ if (isFirstSplitGeneration()) {
+ lastEndTimestamp = getCurrentHybridTime();
+ newSplits =
splitGenerator.generateFullScanSplits(lastEndTimestamp);
+ } else {
+ long startHT = lastEndTimestamp;
+ long endHT = getCurrentHybridTime();
+ newSplits = splitGenerator.generateIncrementalSplits(startHT,
endHT);
+ lastEndTimestamp = endHT;
+ }
+
+ return newSplits;
+ }
+
+ private void assignSplits(List<KuduSourceSplit> splits, Throwable error) {
+ if (error != null) {
+ LOG.error("Failed to enumerate Kudu splits.", error);
+ return;
+ }
+
+ if (splits != null) {
+ unassigned.addAll(splits);
+ }
+ assignSplitsToReaders();
+ }
+
+ private void assignSplitsToReaders() {
+ final Iterator<Integer> awaitingSubtasks =
readersAwaitingSplit.iterator();
+
+ while (awaitingSubtasks.hasNext()) {
+ final int awaitingSubtask = awaitingSubtasks.next();
+
+ // If the reader that requested another split has failed in the
meantime, remove
+ // it from the list of waiting readers
+ if (!context.registeredReaders().containsKey(awaitingSubtask)) {
+ awaitingSubtasks.remove();
+ continue;
+ }
+
+ final Optional<KuduSourceSplit> nextSplit = getNextSplit();
+ if (nextSplit.isPresent()) {
+ context.assignSplit(nextSplit.get(), awaitingSubtask);
+ awaitingSubtasks.remove();
+ pending.add(nextSplit.get());
+ } else {
+ break;
+ }
+ }
+ }
+
+ private Optional<KuduSourceSplit> getNextSplit() {
Review Comment:
This method is duplicated in `KuduSourceSplitReader`. I think it makes sense
to create a `source.utils` package and separate this a utility class,
`KuduSplitRetriever` or something, and the method can take a `Collection` and
retrieve the next split from that, so it can be used here and in
`KuduSourceSplitReader` as well. And if we have a `utils` package, IMO
`KuduSplitGenerator` should live there as well, it seems more appropriate to me.
Furthermore, this is probably arguable but IMO the call side of method does
not really utilizes the benefits of `Optional`, so we can simply return `null`
or the actual element, and have a `null` check inside `assignSplitsToReaders`,
and we csan spare 2 `.get()` calls. Feel free to incorporate this, or leave it
as is.
##########
flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/enumerator/KuduSourceEnumerator.java:
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.config.ContinuousBoundingSettings;
+import org.apache.flink.connector.kudu.source.split.KuduSourceSplit;
+import org.apache.flink.connector.kudu.source.split.SplitFinishedEvent;
+
+import org.apache.kudu.util.HybridTimeUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The Kudu source enumerator is responsible for periodically discovering and
assigning new splits
+ * when possible.
+ *
+ * <p>To provide CDC-like functionality, the enumeration works as follows:
Initially, we perform a
+ * snapshot read of the table and mark the snapshot time as t0. From that
point onward, we perform
+ * differential scans in the time intervals t0 - t1, t1 - t2, and so on.
+ *
+ * <p>This approach means that new splits can only be enumerated once the
current time range is
+ * fully processed.
+ *
+ * <p>The process is controlled as follows:
+ *
+ * <ul>
+ * <li>Once a set of splits is enumerated for a time range, we track:
+ * <ul>
+ * <li><b>Unassigned splits</b>: Discovered but not yet assigned to
readers.
+ * <li><b>Pending splits</b>: Assigned but not yet fully processed.
+ * </ul>
+ * <li>A new set of splits is generated only when there are no remaining
unassigned or pending
+ * splits.
+ * </ul>
+ */
+public class KuduSourceEnumerator
+ implements SplitEnumerator<KuduSourceSplit, KuduSourceEnumeratorState>
{
+ private static final Logger LOG =
LoggerFactory.getLogger(KuduSourceEnumerator.class);
+
+ private final SplitEnumeratorContext<KuduSourceSplit> context;
+ private final List<Integer> readersAwaitingSplit;
+ private final KuduSplitGenerator splitGenerator;
+ private final ContinuousBoundingSettings continuousBoundingSettings;
+
+ private long lastEndTimestamp;
+ private final List<KuduSourceSplit> unassigned;
+ private final List<KuduSourceSplit> pending;
+
+ public KuduSourceEnumerator(
+ KuduTableInfo tableInfo,
+ KuduReaderConfig readerConfig,
+ ContinuousBoundingSettings continuousBoundingSettings,
+ SplitEnumeratorContext<KuduSourceSplit> context) {
+ this(
+ tableInfo,
+ readerConfig,
+ continuousBoundingSettings,
+ context,
+ KuduSourceEnumeratorState.empty());
+ }
+
+ public KuduSourceEnumerator(
+ KuduTableInfo tableInfo,
+ KuduReaderConfig readerConfig,
+ ContinuousBoundingSettings continuousBoundingSettings,
+ SplitEnumeratorContext<KuduSourceSplit> context,
+ KuduSourceEnumeratorState enumState) {
+ this.continuousBoundingSettings = continuousBoundingSettings;
+ 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();
+ }
+
+ @Override
+ public void start() {
+ if (continuousBoundingSettings.getBoundedness() ==
Boundedness.CONTINUOUS_UNBOUNDED
+ && Objects.nonNull(continuousBoundingSettings.getPeriod())) {
+ context.callAsync(
+ () -> enumerateNewSplits(() -> pending.isEmpty() &&
unassigned.isEmpty()),
+ this::assignSplits,
+ 0,
+ continuousBoundingSettings.getPeriod().toMillis());
+ } else {
+ List<KuduSourceSplit> splits = enumerateNewSplits(() -> true);
+ if (splits != null) {
+ unassigned.addAll(enumerateNewSplits(() -> true));
+ assignSplitsToReaders();
+ }
Review Comment:
Why do we call `enumerateNewSplits(() -> true)` @L123 again? IMO this could
be simplified to:
```java
List<KuduSourceSplit> splits = enumerateNewSplits(() -> true);
assignSplits(splits, null);
```
##########
flink-connector-kudu/src/main/java/org/apache/flink/connector/kudu/source/KuduSource.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.config.BoundednessSettings;
+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;
+
+/**
+ * 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.
+ *
+ * <p>The behavior depends on the {@link BoundednessSettings}:
+ *
+ * <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 BoundednessSettings} - Defines the boundedness and 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 BoundednessSettings boundednessSettings;
+ private final RowResultConverter<OUT> rowResultConverter;
+
+ private final Configuration configuration;
+
+ KuduSource(
+ KuduReaderConfig readerConfig,
+ KuduTableInfo tableInfo,
+ BoundednessSettings boundednessSettings,
+ RowResultConverter<OUT> rowResultConverter) {
+ this.tableInfo = tableInfo;
+ this.readerConfig = readerConfig;
+ this.rowResultConverter = rowResultConverter;
+ this.boundednessSettings = boundednessSettings;
+ this.configuration = new Configuration();
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return boundednessSettings.getBoundedness();
+ }
+
+ @Override
+ public SplitEnumerator<KuduSourceSplit, KuduSourceEnumeratorState>
createEnumerator(
+ SplitEnumeratorContext<KuduSourceSplit> enumContext) {
+ return new KuduSourceEnumerator(tableInfo, readerConfig,
boundednessSettings, enumContext);
+ }
+
+ @Override
+ public SplitEnumerator<KuduSourceSplit, KuduSourceEnumeratorState>
restoreEnumerator(
+ SplitEnumeratorContext<KuduSourceSplit> enumContext,
+ KuduSourceEnumeratorState checkpoint)
+ throws Exception {
+ return new KuduSourceEnumerator(
+ tableInfo, readerConfig, boundednessSettings, 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(
Review Comment:
This should be `return new KuduSourceReader<>(...);`, which should be
possible after the changes I suggested at the `KuduSourceReader` ctor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
