leonardBang commented on a change in pull request #18547:
URL: https://github.com/apache/flink/pull/18547#discussion_r805161181
##########
File path:
flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/testsuites/SourceTestSuiteBase.java
##########
@@ -86,6 +92,8 @@
public abstract class SourceTestSuiteBase<T> {
private static final Logger LOG =
LoggerFactory.getLogger(SourceTestSuiteBase.class);
+ static ExecutorService executorService =
+
Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
Review comment:
The number for fixed threads is too expensive for tests.
##########
File path:
flink-test-utils-parent/flink-connector-test-utils/src/test/java/org/apache/flink/connector/testframe/utils/CollectIteratorAssertTest.java
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.testframe.utils;
+
+import org.apache.flink.streaming.api.CheckpointingMode;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static
org.apache.flink.connector.testframe.utils.CollectIteratorAssertions.assertThat;
+
+/** Unit test for {@link CollectIteratorAssertTest}. */
Review comment:
```suggestion
/** Unit tests for {@link CollectIteratorAssertTest}. */
```
##########
File path:
flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/testsuites/SourceTestSuiteBase.java
##########
@@ -400,6 +413,37 @@ protected JobClient submitJob(StreamExecutionEnvironment
env, String jobName) th
stream.getExecutionEnvironment().getCheckpointConfig());
}
+ /**
+ * Compare the test data with the result.
+ *
+ * <p>If the source is bounded, limit should be null.
+ *
+ * @param resultIterator the data read from the job
+ * @param testData the test data
+ * @param semantic the supported semantic, see {@link CheckpointingMode}
+ * @param limit expected number of the data to read from the job
+ */
+ private void checkResultBySemantic(
Review comment:
```suggestion
private void checkResultWithSemantic(
```
##########
File path:
flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/utils/CollectIteratorAssert.java
##########
@@ -0,0 +1,257 @@
+/*
+ * 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.testframe.utils;
+
+import org.apache.flink.streaming.api.CheckpointingMode;
+
+import org.assertj.core.api.AbstractAssert;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+
+/** @param <T> */
Review comment:
Please add reasonable java document
##########
File path:
flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/utils/CollectIteratorAssert.java
##########
@@ -0,0 +1,257 @@
+/*
+ * 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.testframe.utils;
+
+import org.apache.flink.streaming.api.CheckpointingMode;
+
+import org.assertj.core.api.AbstractAssert;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+
+/** @param <T> */
+public class CollectIteratorAssert<T>
+ extends AbstractAssert<CollectIteratorAssert<T>, Iterator<T>> {
+
+ private static final int UNSET = -1;
+ private final Iterator<T> collectorIterator;
+ private final List<RecordsFromSplit<T>> recordsFromSplits = new
ArrayList<>();
+ private int totalNumRecords;
+ private Integer limit = null;
+
+ protected CollectIteratorAssert(Iterator<T> collectorIterator) {
+ super(collectorIterator, CollectIteratorAssert.class);
+ this.collectorIterator = collectorIterator;
+ }
+
+ public CollectIteratorAssert<T> withNumRecordsLimit(int limit) {
+ this.limit = limit;
+ return this;
+ }
+
+ public void matchesRecordsFromSource(
+ List<List<T>> recordsBySplitsFromSource, CheckpointingMode
semantic) {
+ for (List<T> recordsFromSplit : recordsBySplitsFromSource) {
+ recordsFromSplits.add(new RecordsFromSplit<>(recordsFromSplit));
+ totalNumRecords += recordsFromSplit.size();
+ }
+
+ if (limit != null && limit > totalNumRecords) {
+ throw new IllegalArgumentException(
+ "Limit validation size should be less than total number of
records from source");
+ }
+
+ switch (semantic) {
+ case AT_LEAST_ONCE:
+ matchAtLeastOnce(collectorIterator, recordsFromSplits);
+ break;
+ case EXACTLY_ONCE:
+ matchExactlyOnce(collectorIterator, recordsFromSplits);
+ break;
+ default:
+ throw new IllegalArgumentException(
+ String.format("Unrecognized semantic \"%s\"",
semantic));
+ }
+ }
+
+ private void matchAtLeastOnce(
+ Iterator<T> resultIterator, List<RecordsFromSplit<T>>
recordsFromSplits) {
+ List<T> duplicateRead = new LinkedList<>();
Review comment:
how about`duplicatedRecords` or `recordsWithAtLeastOnceSemantic` ? And
this method haven't unit test to cover in this PR?
##########
File path:
flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/utils/CollectIteratorAssertions.java
##########
@@ -0,0 +1,28 @@
+/*
+ * 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.testframe.utils;
+
+import java.util.Iterator;
+
+/** */
+public class CollectIteratorAssertions {
Review comment:
Do we really need this class ?
##########
File path:
flink-test-utils-parent/flink-connector-test-utils/src/test/java/org/apache/flink/connector/testframe/utils/CollectIteratorAssertTest.java
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.testframe.utils;
+
+import org.apache.flink.streaming.api.CheckpointingMode;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static
org.apache.flink.connector.testframe.utils.CollectIteratorAssertions.assertThat;
+
+/** Unit test for {@link CollectIteratorAssertTest}. */
+public class CollectIteratorAssertTest {
+ @Nested
+ class MultipleSplitDataMatcherTest {
+ private final List<String> splitA = Arrays.asList("alpha", "beta",
"gamma");
+ private final List<String> splitB = Arrays.asList("one", "two",
"three");
+ private final List<String> splitC = Arrays.asList("1", "2", "3");
+ private final List<List<String>> testDataCollection =
Arrays.asList(splitA, splitB, splitC);
+
+ @Test
+ public void testPositiveCase() {
Review comment:
```suggestion
public void testDataMatcherWithExactlyOnceSemantic() {
```
Add we can also add ` public void testDataMatcherWithAtLeastOnceSemantic() {`
--
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]
