StephanEwen commented on a change in pull request #12122: URL: https://github.com/apache/flink/pull/12122#discussion_r425072701
########## File path: flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/CoordinatedSourceITCase.java ########## @@ -0,0 +1,63 @@ +/* + 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.common.accumulators.ListAccumulator; +import org.apache.flink.api.connector.source.Boundedness; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.connector.base.source.reader.mocks.MockBaseSource; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.functions.sink.RichSinkFunction; +import org.apache.flink.test.util.AbstractTestBase; + +import org.junit.Test; + +import java.util.List; +import java.util.SortedSet; +import java.util.TreeSet; + +import static org.junit.Assert.assertEquals; + +public class CoordinatedSourceITCase extends AbstractTestBase { + + @Test + public void testEnumeratorReaderCommunication() throws Exception { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + MockBaseSource source = new MockBaseSource(2, 10, Boundedness.BOUNDED); + DataStream<Integer> stream = env.continuousSource(source, "TestingSource"); + stream.addSink(new RichSinkFunction<Integer>() { + @Override + public void open(Configuration parameters) throws Exception { + getRuntimeContext().addAccumulator("result", new ListAccumulator<Integer>()); + } + + @Override + public void invoke(Integer value, Context context) throws Exception { + getRuntimeContext().getAccumulator("result").add(value); + } + }); + List<Integer> result = env.execute().getAccumulatorResult("result"); + SortedSet<Integer> resultSet = new TreeSet<>(result); Review comment: I think it would be nice to compare the entire list. That way we would make sure we don't have accidental losses / duplicates cancelling each other out. And, when the test fails, we immediately see what happens, when the `assertEquals` statement prints the full list (you can see if something was duplicated). ########## File path: flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java ########## @@ -81,6 +82,8 @@ Licensed to the Apache Software Foundation (ASF) under one /** The last element to ensure it is fully handled. */ private SplitsRecordIterator<E> splitIter; + private volatile boolean noMoreSplitsAssignment; Review comment: Does this need to be volatile? Isn't it updated always by the same thread (mailbox thread)? ########## File path: flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/mocks/MockBaseSource.java ########## @@ -115,12 +115,17 @@ public int getVersion() { @Override public byte[] serialize(List<MockSourceSplit> obj) throws IOException { - return SerializationUtils.serialize(obj.toArray()); + return InstantiationUtil.serializeObject(obj.toArray()); } @Override public List<MockSourceSplit> deserialize(int version, byte[] serialized) throws IOException { - MockSourceSplit[] splitArray = SerializationUtils.deserialize(serialized); + MockSourceSplit[] splitArray; + try { + splitArray = InstantiationUtil.deserializeObject(serialized, getClass().getClassLoader()); + } catch (ClassNotFoundException e) { + throw new RuntimeException("Failed to deserialize the source split."); Review comment: How about throwing `IOException` here? Other parts of the code typically do it like that. The checked IOException is supposed to indicate failed deserialization (for any reason), so it makes sense to throw that. ########## File path: flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/mocks/MockBaseSource.java ########## @@ -0,0 +1,128 @@ +/* + 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.commons.lang3.SerializationUtils; +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.api.connector.source.mocks.MockSourceSplit; +import org.apache.flink.api.connector.source.mocks.MockSourceSplitSerializer; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds; +import org.apache.flink.connector.base.source.reader.SourceReaderOptions; +import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue; +import org.apache.flink.connector.base.source.reader.synchronization.FutureNotifier; +import org.apache.flink.core.io.SimpleVersionedSerializer; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +/** + * A {@link Source} class for unit test of base implementation of source connector. + */ +public class MockBaseSource implements Source<Integer, MockSourceSplit, List<MockSourceSplit>> { + private final int numSplits; + private final int numRecordsPerSplit; + private final Boundedness boundedness; + + public MockBaseSource(int numSplits, int numRecordsPerSplit, Boundedness boundedness) { + this.numSplits = numSplits; + this.numRecordsPerSplit = numRecordsPerSplit; + this.boundedness = boundedness; + } + + @Override + public Boundedness getBoundedness() { + return boundedness; + } + + @Override + public SourceReader<Integer, MockSourceSplit> createReader(SourceReaderContext readerContext) { + FutureNotifier futureNotifier = new FutureNotifier(); + FutureCompletingBlockingQueue<RecordsWithSplitIds<int[]>> elementsQueue = + new FutureCompletingBlockingQueue<>(futureNotifier); + + Configuration config = new Configuration(); + config.setInteger(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY, 1); + config.setLong(SourceReaderOptions.SOURCE_READER_CLOSE_TIMEOUT, 30000L); + return new MockSourceReader( + futureNotifier, + elementsQueue, + () -> new MockSplitReader(2, true, true), + config, + readerContext); + } + + @Override + public SplitEnumerator<MockSourceSplit, List<MockSourceSplit>> createEnumerator( + SplitEnumeratorContext<MockSourceSplit> enumContext) { + List<MockSourceSplit> splits = new ArrayList<>(); + for (int i = 0; i < numSplits; i++) { + int endIndex = boundedness == Boundedness.BOUNDED ? numRecordsPerSplit : Integer.MAX_VALUE; + MockSourceSplit split = new MockSourceSplit(i, 0, endIndex); + for (int j = 0; j < numRecordsPerSplit; j++) { + split.addRecord(i * numRecordsPerSplit + j); + } + splits.add(split); + } + return new MockSplitEnumerator(splits, enumContext); + } + + @Override + public SplitEnumerator<MockSourceSplit, List<MockSourceSplit>> restoreEnumerator( + SplitEnumeratorContext<MockSourceSplit> enumContext, + List<MockSourceSplit> checkpoint) throws IOException { + FutureNotifier futureNotifier = new FutureNotifier(); Review comment: The FutureNotifier and the queue seem not needed here. ---------------------------------------------------------------- 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
