You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2020/11/10 14:04:52 UTC
[flink] branch master updated (9b12c15 -> d00941c)
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 9b12c15 [FLINK-19436][tests] Properly shutdown cluster in e2e tests
new 4942e8f [refactor] Move Source Reader Test Utils classes from 'flink-core' test jar to 'flink-connector-test-utils'
new d00941c [FLINK-20063][connector files] FileSourceReader request only a split if it doesn't have one already
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../base/source/reader/SourceReaderBase.java | 11 +++
.../base/source/reader/SourceReaderBaseTest.java | 4 +-
flink-connectors/flink-connector-files/pom.xml | 7 ++
.../connector/file/src/impl/FileSourceReader.java | 5 +-
.../file/src/impl/FileSourceReaderTest.java | 81 ++++++++++++++++++++
.../source/lib/NumberSequenceSourceTest.java | 86 +++++++++++++++++++++-
.../source/reader}/TestingReaderContext.java | 12 ++-
.../source/reader}/TestingReaderOutput.java | 2 +-
8 files changed, 199 insertions(+), 9 deletions(-)
create mode 100644 flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/FileSourceReaderTest.java
rename {flink-core/src/test/java/org/apache/flink/api/connector/source/mocks => flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader}/TestingReaderContext.java (91%)
rename {flink-core/src/test/java/org/apache/flink/api/connector/source/mocks => flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader}/TestingReaderOutput.java (97%)
[flink] 01/02: [refactor] Move Source Reader Test Utils classes
from 'flink-core' test jar to 'flink-connector-test-utils'
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4942e8f359276a57e59f39949abfc3e9e2eba569
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Mon Nov 9 20:35:17 2020 +0100
[refactor] Move Source Reader Test Utils classes from 'flink-core' test jar to 'flink-connector-test-utils'
---
.../base/source/reader/SourceReaderBaseTest.java | 4 +-
.../source/lib/NumberSequenceSourceTest.java | 86 +++++++++++++++++++++-
.../source/reader}/TestingReaderContext.java | 12 ++-
.../source/reader}/TestingReaderOutput.java | 2 +-
4 files changed, 96 insertions(+), 8 deletions(-)
diff --git a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/SourceReaderBaseTest.java b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/SourceReaderBaseTest.java
index 5c8d9d3..f07306b 100644
--- a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/SourceReaderBaseTest.java
+++ b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/SourceReaderBaseTest.java
@@ -21,8 +21,6 @@ package org.apache.flink.connector.base.source.reader;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.SourceReader;
import org.apache.flink.api.connector.source.mocks.MockSourceSplit;
-import org.apache.flink.api.connector.source.mocks.TestingReaderContext;
-import org.apache.flink.api.connector.source.mocks.TestingReaderOutput;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.source.reader.mocks.MockSourceReader;
import org.apache.flink.connector.base.source.reader.mocks.MockSplitReader;
@@ -34,6 +32,8 @@ 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.base.source.reader.synchronization.FutureCompletingBlockingQueue;
import org.apache.flink.connector.testutils.source.reader.SourceReaderTestBase;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput;
import org.apache.flink.core.io.InputStatus;
import org.junit.Rule;
diff --git a/flink-core/src/test/java/org/apache/flink/api/connector/source/lib/NumberSequenceSourceTest.java b/flink-core/src/test/java/org/apache/flink/api/connector/source/lib/NumberSequenceSourceTest.java
index 48cda2d..518ce86 100644
--- a/flink-core/src/test/java/org/apache/flink/api/connector/source/lib/NumberSequenceSourceTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/connector/source/lib/NumberSequenceSourceTest.java
@@ -18,13 +18,20 @@
package org.apache.flink.api.connector.source.lib;
+import org.apache.flink.api.common.eventtime.Watermark;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceEvent;
+import org.apache.flink.api.connector.source.SourceOutput;
import org.apache.flink.api.connector.source.SourceReader;
-import org.apache.flink.api.connector.source.mocks.TestingReaderContext;
-import org.apache.flink.api.connector.source.mocks.TestingReaderOutput;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.junit.Test;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -86,6 +93,79 @@ public class NumberSequenceSourceTest {
private static SourceReader<Long, NumberSequenceSource.NumberSequenceSplit> createReader() {
// the arguments passed in the source constructor matter only to the enumerator
- return new NumberSequenceSource(0L, 0L).createReader(new TestingReaderContext());
+ return new NumberSequenceSource(0L, 0L).createReader(new DummyReaderContext());
+ }
+
+ // ------------------------------------------------------------------------
+ // test utils / mocks
+ //
+ // the "flink-connector-test-utils module has proper mocks and utils,
+ // but cannot be used here, because it would create a cyclic dependency.
+ // ------------------------------------------------------------------------
+
+ private static final class DummyReaderContext implements SourceReaderContext {
+
+ @Override
+ public MetricGroup metricGroup() {
+ return new UnregisteredMetricsGroup();
+ }
+
+ @Override
+ public Configuration getConfiguration() {
+ return new Configuration();
+ }
+
+ @Override
+ public String getLocalHostName() {
+ return "localhost";
+ }
+
+ @Override
+ public int getIndexOfSubtask() {
+ return 0;
+ }
+
+ @Override
+ public void sendSplitRequest() {}
+
+ @Override
+ public void sendSourceEventToCoordinator(SourceEvent sourceEvent) {}
+ }
+
+ private static final class TestingReaderOutput<E> implements ReaderOutput<E> {
+
+ private final ArrayList<E> emittedRecords = new ArrayList<>();
+
+ @Override
+ public void collect(E record) {
+ emittedRecords.add(record);
+ }
+
+ @Override
+ public void collect(E record, long timestamp) {
+ collect(record);
+ }
+
+ @Override
+ public void emitWatermark(Watermark watermark) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void markIdle() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SourceOutput<E> createOutputForSplit(String splitId) {
+ return this;
+ }
+
+ @Override
+ public void releaseOutputForSplit(String splitId) {}
+
+ public ArrayList<E> getEmittedRecords() {
+ return emittedRecords;
+ }
}
}
diff --git a/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/TestingReaderContext.java b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderContext.java
similarity index 91%
rename from flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/TestingReaderContext.java
rename to flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderContext.java
index 98aad54..1fda64a 100644
--- a/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/TestingReaderContext.java
+++ b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderContext.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.api.connector.source.mocks;
+package org.apache.flink.connector.testutils.source.reader;
import org.apache.flink.api.connector.source.SourceEvent;
import org.apache.flink.api.connector.source.SourceReaderContext;
@@ -38,6 +38,8 @@ public class TestingReaderContext implements SourceReaderContext {
private final ArrayList<SourceEvent> sentEvents = new ArrayList<>();
+ private int numSplitRequests;
+
public TestingReaderContext() {
this(new Configuration());
}
@@ -69,7 +71,9 @@ public class TestingReaderContext implements SourceReaderContext {
}
@Override
- public void sendSplitRequest() {}
+ public void sendSplitRequest() {
+ numSplitRequests++;
+ }
@Override
public void sendSourceEventToCoordinator(SourceEvent sourceEvent) {
@@ -78,6 +82,10 @@ public class TestingReaderContext implements SourceReaderContext {
// ------------------------------------------------------------------------
+ public int getNumSplitRequests() {
+ return numSplitRequests;
+ }
+
public List<SourceEvent> getSentEvents() {
return new ArrayList<>(sentEvents);
}
diff --git a/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/TestingReaderOutput.java b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderOutput.java
similarity index 97%
rename from flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/TestingReaderOutput.java
rename to flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderOutput.java
index 1847cc4..5617d4d4 100644
--- a/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/TestingReaderOutput.java
+++ b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderOutput.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.api.connector.source.mocks;
+package org.apache.flink.connector.testutils.source.reader;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.connector.source.ReaderOutput;
[flink] 02/02: [FLINK-20063][connector files] FileSourceReader
request only a split if it doesn't have one already
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d00941c77170b233c9fe599c7fb0003778eb3299
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Mon Nov 9 20:36:56 2020 +0100
[FLINK-20063][connector files] FileSourceReader request only a split if it doesn't have one already
This closes #14004
---
.../base/source/reader/SourceReaderBase.java | 11 +++
flink-connectors/flink-connector-files/pom.xml | 7 ++
.../connector/file/src/impl/FileSourceReader.java | 5 +-
.../file/src/impl/FileSourceReaderTest.java | 81 ++++++++++++++++++++++
4 files changed, 103 insertions(+), 1 deletion(-)
diff --git a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java
index b8dc9ef..7fa548b 100644
--- a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java
+++ b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java
@@ -236,6 +236,17 @@ public abstract class SourceReaderBase<E, T, SplitT extends SourceSplit, SplitSt
splitFetcherManager.close(options.sourceReaderCloseTimeout);
}
+ /**
+ * Gets the number of splits the reads has currently assigned.
+ *
+ * <p>These are the splits that have been added via {@link #addSplits(List)} and have not
+ * yet been finished by returning them from the {@link SplitReader#fetch()} as part of
+ * {@link RecordsWithSplitIds#finishedSplits()}.
+ */
+ public int getNumberOfCurrentlyAssignedSplits() {
+ return splitStates.size();
+ }
+
// -------------------- Abstract method to allow different implementations ------------------
/**
* Handles the finished splits to clean the state if needed.
diff --git a/flink-connectors/flink-connector-files/pom.xml b/flink-connectors/flink-connector-files/pom.xml
index 12ac1a2..b6cb260 100644
--- a/flink-connectors/flink-connector-files/pom.xml
+++ b/flink-connectors/flink-connector-files/pom.xml
@@ -85,6 +85,13 @@ under the License.
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-test-utils</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
</dependencies>
</project>
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileSourceReader.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileSourceReader.java
index d64b226..57d14d6 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileSourceReader.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/FileSourceReader.java
@@ -47,7 +47,10 @@ public final class FileSourceReader<T, SplitT extends FileSourceSplit>
@Override
public void start() {
- context.sendSplitRequest();
+ // we request a split only if we did not get splits during the checkpoint restore
+ if (getNumberOfCurrentlyAssignedSplits() == 0) {
+ context.sendSplitRequest();
+ }
}
@Override
diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/FileSourceReaderTest.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/FileSourceReaderTest.java
new file mode 100644
index 0000000..ef09e0e
--- /dev/null
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/FileSourceReaderTest.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.file.src.impl;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.file.src.FileSourceSplit;
+import org.apache.flink.connector.file.src.reader.TextLineFormat;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
+import org.apache.flink.core.fs.Path;
+
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+import java.util.Collections;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Unit tests for the {@link FileSourceReader}.
+ */
+public class FileSourceReaderTest {
+
+ @ClassRule
+ public static final TemporaryFolder TMP_DIR = new TemporaryFolder();
+
+ @Test
+ public void testRequestSplitWhenNoSplitRestored() throws Exception {
+ final TestingReaderContext context = new TestingReaderContext();
+ final FileSourceReader<String, FileSourceSplit> reader = createReader(context);
+
+ reader.start();
+ reader.close();
+
+ assertEquals(1, context.getNumSplitRequests());
+ }
+
+ @Test
+ public void testNoSplitRequestWhenSplitRestored() throws Exception {
+ final TestingReaderContext context = new TestingReaderContext();
+ final FileSourceReader<String, FileSourceSplit> reader = createReader(context);
+
+ reader.addSplits(Collections.singletonList(createTestFileSplit()));
+ reader.start();
+ reader.close();
+
+ assertEquals(0, context.getNumSplitRequests());
+ }
+
+ private static FileSourceReader<String, FileSourceSplit> createReader(TestingReaderContext context) {
+ return new FileSourceReader<>(
+ context,
+ new StreamFormatAdapter<>(new TextLineFormat()),
+ new Configuration());
+ }
+
+ private static FileSourceSplit createTestFileSplit() throws IOException {
+ return new FileSourceSplit(
+ "test-id",
+ Path.fromLocalFile(TMP_DIR.newFile()),
+ 0L,
+ 0L);
+ }
+}