becketqin commented on a change in pull request #13344:
URL: https://github.com/apache/flink/pull/13344#discussion_r486316183
##########
File path:
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherTask.java
##########
@@ -32,7 +34,7 @@
* @return whether the runnable has successfully finished running.
* @throws InterruptedException when interrupted.
*/
Review comment:
nit: missing `@throws IOException`.
##########
File path:
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/SourceReaderBaseUnitTest.java
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.base.source.reader;
+
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.configuration.Configuration;
+import
org.apache.flink.connector.base.source.reader.mocks.PassThroughRecordEmitter;
+import
org.apache.flink.connector.base.source.reader.mocks.TestingReaderContext;
+import org.apache.flink.connector.base.source.reader.mocks.TestingReaderOutput;
+import
org.apache.flink.connector.base.source.reader.mocks.TestingRecordsWithSplitIds;
+import org.apache.flink.connector.base.source.reader.mocks.TestingSourceSplit;
+import org.apache.flink.connector.base.source.reader.mocks.TestingSplitReader;
+import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
+import
org.apache.flink.connector.base.source.reader.synchronization.FutureNotifier;
+
+import org.junit.Test;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Targeted unit tests for the {@link SourceReaderBase}.
+ */
+public class SourceReaderBaseUnitTest {
+
+ @Test
+ public void recordsWithSplitsNotRecycledWhenRecordsLeft() throws
Exception {
Review comment:
nit: it seems the test methods all start with "test". Should this be
`testRecordsWithSplitsNotRecycledWhenRecordsLeft`?
##########
File path:
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/SourceReaderBaseUnitTest.java
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.base.source.reader;
+
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.configuration.Configuration;
+import
org.apache.flink.connector.base.source.reader.mocks.PassThroughRecordEmitter;
+import
org.apache.flink.connector.base.source.reader.mocks.TestingReaderContext;
+import org.apache.flink.connector.base.source.reader.mocks.TestingReaderOutput;
+import
org.apache.flink.connector.base.source.reader.mocks.TestingRecordsWithSplitIds;
+import org.apache.flink.connector.base.source.reader.mocks.TestingSourceSplit;
+import org.apache.flink.connector.base.source.reader.mocks.TestingSplitReader;
+import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
+import
org.apache.flink.connector.base.source.reader.synchronization.FutureNotifier;
+
+import org.junit.Test;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Targeted unit tests for the {@link SourceReaderBase}.
+ */
+public class SourceReaderBaseUnitTest {
Review comment:
Should we merge this class with `SourceReaderBaseTest`?
##########
File path:
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/mocks/TestingReaderOutput.java
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.base.source.reader.mocks;
+
+import org.apache.flink.api.common.eventtime.Watermark;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceOutput;
+
+import java.util.ArrayList;
+
+/**
+ * A {@code ReaderOutput} for testing that collects the emitted records.
+ */
+public class TestingReaderOutput<E> implements ReaderOutput<E> {
Review comment:
I am wondering if some of these testing implementations can be merged
with the existing mock classes. It seems some of the functionalities can be
shared.
##########
File path:
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java
##########
@@ -99,67 +106,94 @@ public SourceReaderBase(
this.splitFetcherManager = splitFetcherManager;
this.recordEmitter = recordEmitter;
this.splitStates = new HashMap<>();
- this.splitIter = null;
this.options = new SourceReaderOptions(config);
this.config = config;
this.context = context;
this.noMoreSplitsAssignment = false;
}
@Override
- public void start() {
-
- }
+ public void start() {}
@Override
public InputStatus pollNext(ReaderOutput<T> output) throws Exception {
splitFetcherManager.checkErrors();
- // poll from the queue if the last element was successfully
handled. Otherwise
- // just pass the last element again.
- RecordsWithSplitIds<E> recordsWithSplitId = null;
- boolean newFetch = splitIter == null || !splitIter.hasNext();
- if (newFetch) {
- recordsWithSplitId = elementsQueue.poll();
+
+ // make sure we have a fetch we are working on, or move to the
next
Review comment:
Thanks for the refactoring, the logic is much clearer now.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
