You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by jq...@apache.org on 2020/11/24 09:36:54 UTC
[flink] branch FLINK-19743 created (now 4cb7775)
This is an automated email from the ASF dual-hosted git repository.
jqin pushed a change to branch FLINK-19743
in repository https://gitbox.apache.org/repos/asf/flink.git.
at 4cb7775 [FLINK-19743][connectors/common] Report numRecordsIn and numRecordsInPerSecond in SourceReaderBase.
This branch includes the following new commits:
new 560f61e [FLINK-19743][connectors/common] Add metric definitions of Source. (FLIP-33) - Expose the Source metrics via SourceMetricGroup and SourceReaderContext. - Let SourceOperator report the WatermarkLag.
new 4cb7775 [FLINK-19743][connectors/common] Report numRecordsIn and numRecordsInPerSecond in SourceReaderBase.
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.
[flink] 02/02: [FLINK-19743][connectors/common] Report numRecordsIn
and numRecordsInPerSecond in SourceReaderBase.
Posted by jq...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jqin pushed a commit to branch FLINK-19743
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4cb77755791ae9885e969f595011bb8f9d8a6d88
Author: Jiangjie (Becket) Qin <ji...@alibaba-inc.com>
AuthorDate: Fri Nov 13 11:28:21 2020 +0800
[FLINK-19743][connectors/common] Report numRecordsIn and numRecordsInPerSecond in SourceReaderBase.
---
.../base/source/reader/SourceReaderBase.java | 8 +++-
.../base/source/reader/SourceReaderBaseTest.java | 48 ++++++++++++++++++----
2 files changed, 47 insertions(+), 9 deletions(-)
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 0a7a742..4355fb3 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
@@ -29,6 +29,7 @@ import org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager
import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.metrics.Counter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -77,7 +78,10 @@ public abstract class SourceReaderBase<E, T, SplitT extends SourceSplit, SplitSt
protected final Configuration config;
/** The context of this source reader. */
- protected SourceReaderContext context;
+ protected final SourceReaderContext context;
+
+ /** A counter that records the number of records consumed by this source reader. */
+ private final Counter numRecordsInCounter;
/** The latest fetched batch of records-by-split from the split reader. */
@Nullable private RecordsWithSplitIds<E> currentFetch;
@@ -100,6 +104,7 @@ public abstract class SourceReaderBase<E, T, SplitT extends SourceSplit, SplitSt
this.options = new SourceReaderOptions(config);
this.config = config;
this.context = context;
+ this.numRecordsInCounter = context.metricGroup().getNumRecordsInCounter();
this.noMoreSplitsAssignment = false;
}
@@ -123,6 +128,7 @@ public abstract class SourceReaderBase<E, T, SplitT extends SourceSplit, SplitSt
final E record = recordsWithSplitId.nextRecordFromSplit();
if (record != null) {
// emit the record.
+ numRecordsInCounter.inc();
recordEmitter.emitRecord(record, currentSplitOutput, currentSplitContext.state);
LOG.trace("Emitted record: {}", record);
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 5d5bc20..9ab014d 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
@@ -19,7 +19,9 @@
package org.apache.flink.connector.base.source.reader;
import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.ReaderOutput;
import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.api.connector.source.SourceSplit;
import org.apache.flink.api.connector.source.mocks.MockSourceSplit;
import org.apache.flink.configuration.Configuration;
@@ -90,7 +92,7 @@ public class SourceReaderBaseTest extends SourceReaderTestBase<MockSourceSplit>
public void close() {}
},
getConfig(),
- null)) {
+ new TestingReaderContext())) {
ValidatingSourceOutput output = new ValidatingSourceOutput();
reader.addSplits(Collections.singletonList(getSplit(0,
NUM_RECORDS_PER_SPLIT,
@@ -130,6 +132,25 @@ public class SourceReaderBaseTest extends SourceReaderTestBase<MockSourceSplit>
}
@Test
+ public void testMetrics() throws Exception {
+ final TestingRecordsWithSplitIds<String> records = new TestingRecordsWithSplitIds<>("test-split", "value1", "value2");
+ final SourceReaderContext context = new TestingReaderContext();
+ final SourceReader<String, ?> reader = createReaderAndAwaitAvailable(
+ "test-split",
+ records,
+ new PassThroughRecordEmitter<>(),
+ context);
+
+ ReaderOutput<String> readerOutput = new TestingReaderOutput<>();
+ assertEquals(0, context.metricGroup().getNumRecordsInCounter().getCount());
+ reader.pollNext(readerOutput);
+ assertEquals(1, context.metricGroup().getNumRecordsInCounter().getCount());
+ reader.pollNext(readerOutput);
+ assertEquals(InputStatus.NOTHING_AVAILABLE, reader.pollNext(readerOutput));
+ assertEquals(2, context.metricGroup().getNumRecordsInCounter().getCount());
+ }
+
+ @Test
public void testMultipleSplitsWithDifferentFinishingMoments() throws Exception {
FutureCompletingBlockingQueue<RecordsWithSplitIds<int[]>> elementsQueue =
new FutureCompletingBlockingQueue<>();
@@ -142,7 +163,7 @@ public class SourceReaderBaseTest extends SourceReaderTestBase<MockSourceSplit>
elementsQueue,
() -> mockSplitReader,
getConfig(),
- null);
+ new TestingReaderContext());
reader.start();
@@ -176,7 +197,7 @@ public class SourceReaderBaseTest extends SourceReaderTestBase<MockSourceSplit>
elementsQueue,
() -> mockSplitReader,
getConfig(),
- null);
+ new TestingReaderContext());
reader.start();
@@ -211,7 +232,7 @@ public class SourceReaderBaseTest extends SourceReaderTestBase<MockSourceSplit>
elementsQueue,
splitFetcherManager,
getConfig(),
- null);
+ new TestingReaderContext());
// Create and add a split that only contains one record
final MockSourceSplit split = new MockSourceSplit(0, 0, 1);
@@ -236,7 +257,7 @@ public class SourceReaderBaseTest extends SourceReaderTestBase<MockSourceSplit>
elementsQueue,
() -> mockSplitReader,
getConfig(),
- null);
+ new TestingReaderContext());
}
@Override
@@ -281,16 +302,27 @@ public class SourceReaderBaseTest extends SourceReaderTestBase<MockSourceSplit>
private static <E> SourceReader<E, ?> createReaderAndAwaitAvailable(
final String splitId,
final RecordsWithSplitIds<E> records) throws Exception {
+ return createReaderAndAwaitAvailable(
+ splitId, records, new PassThroughRecordEmitter<>(), new TestingReaderContext());
+ }
+
+ private static <E> SourceReader<E, ?> createReaderAndAwaitAvailable(
+ final String splitId,
+ final RecordsWithSplitIds<E> records,
+ final RecordEmitter<E, E, TestingSourceSplit> recordEmitter,
+ final SourceReaderContext context) throws Exception {
final FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue =
new FutureCompletingBlockingQueue<>();
final SourceReader<E, TestingSourceSplit> reader = new SingleThreadMultiplexSourceReaderBase<E, E, TestingSourceSplit, TestingSourceSplit>(
elementsQueue,
- () -> new TestingSplitReader<>(records),
- new PassThroughRecordEmitter<>(),
+ new SingleThreadFetcherManager<>(
+ elementsQueue,
+ () -> new TestingSplitReader<>(records)),
+ recordEmitter,
new Configuration(),
- new TestingReaderContext()) {
+ context) {
@Override
public void notifyCheckpointComplete(long checkpointId) {}
[flink] 01/02: [FLINK-19743][connectors/common] Add metric
definitions of Source. (FLIP-33) - Expose the Source metrics via
SourceMetricGroup and SourceReaderContext. - Let SourceOperator report the
WatermarkLag.
Posted by jq...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jqin pushed a commit to branch FLINK-19743
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 560f61e33a8307b83a989c5e6e656c4f45edb092
Author: Jiangjie (Becket) Qin <ji...@alibaba-inc.com>
AuthorDate: Thu Oct 22 08:21:21 2020 +0800
[FLINK-19743][connectors/common] Add metric definitions of Source. (FLIP-33)
- Expose the Source metrics via SourceMetricGroup and SourceReaderContext.
- Let SourceOperator report the WatermarkLag.
---
.../file/src/FileSourceHeavyThroughputTest.java | 8 +--
.../api/connector/source/SourceReaderContext.java | 4 +-
.../source/metrics/SourceMetricGroup.java | 54 +++++++++++++++
.../source/lib/NumberSequenceSourceTest.java | 26 +++++++-
.../source/mocks/MockSourceMetricGroup.java | 50 ++++++++++++++
.../connector/source/mocks/MockSourceReader.java | 12 +++-
.../metrics/groups/SourceMetricGroupImpl.java | 76 ++++++++++++++++++++++
flink-streaming-java/pom.xml | 7 ++
.../streaming/api/operators/SourceOperator.java | 58 +++++++++++++++--
.../source/NoOpTimestampsAndWatermarks.java | 6 ++
.../source/ProgressiveTimestampsAndWatermarks.java | 14 +++-
.../operators/source/TimestampsAndWatermarks.java | 7 ++
.../operators/source/WatermarkToDataOutput.java | 4 ++
.../api/operators/SourceOperatorTest.java | 27 ++++++++
.../api/operators/source/CollectingDataOutput.java | 4 +-
.../operators/source/TestingSourceOperator.java | 6 +-
...tStreamTaskChainedSourcesCheckpointingTest.java | 8 +--
.../runtime/tasks/MultipleInputStreamTaskTest.java | 15 ++---
.../tasks/SourceOperatorStreamTaskTest.java | 3 +-
.../source/reader/TestingReaderContext.java | 7 +-
.../source/reader/TestingSourceMetricGroup.java | 48 ++++++++++++++
21 files changed, 406 insertions(+), 38 deletions(-)
diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/FileSourceHeavyThroughputTest.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/FileSourceHeavyThroughputTest.java
index 9a43643..43a1efe 100644
--- a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/FileSourceHeavyThroughputTest.java
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/FileSourceHeavyThroughputTest.java
@@ -26,15 +26,15 @@ 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.SourceReaderContext;
+import org.apache.flink.api.connector.source.metrics.SourceMetricGroup;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.src.reader.SimpleStreamFormat;
import org.apache.flink.connector.file.src.reader.StreamFormat;
import org.apache.flink.connector.file.src.testutils.TestingFileSystem;
+import org.apache.flink.connector.testutils.source.reader.TestingSourceMetricGroup;
import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.io.InputStatus;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.junit.After;
import org.junit.Test;
@@ -186,8 +186,8 @@ public class FileSourceHeavyThroughputTest {
private static final class NoOpReaderContext implements SourceReaderContext {
@Override
- public MetricGroup metricGroup() {
- return new UnregisteredMetricsGroup();
+ public SourceMetricGroup metricGroup() {
+ return new TestingSourceMetricGroup();
}
@Override
diff --git a/flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReaderContext.java b/flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReaderContext.java
index f1ea255..2c5c696 100644
--- a/flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReaderContext.java
+++ b/flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReaderContext.java
@@ -19,8 +19,8 @@
package org.apache.flink.api.connector.source;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.connector.source.metrics.SourceMetricGroup;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.metrics.MetricGroup;
/**
* The class that expose some context from runtime to the {@link SourceReader}.
@@ -31,7 +31,7 @@ public interface SourceReaderContext {
/**
* @return The metric group this source belongs to.
*/
- MetricGroup metricGroup();
+ SourceMetricGroup metricGroup();
/**
* Gets the configuration with which Flink was started.
diff --git a/flink-core/src/main/java/org/apache/flink/api/connector/source/metrics/SourceMetricGroup.java b/flink-core/src/main/java/org/apache/flink/api/connector/source/metrics/SourceMetricGroup.java
new file mode 100644
index 0000000..d6fc667
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/connector/source/metrics/SourceMetricGroup.java
@@ -0,0 +1,54 @@
+/*
+ * 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.api.connector.source.metrics;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
+
+/**
+ * An interface that defines the metric names of the Source (FLIP-27). The interface
+ * also exposes the methods for the predefined metrics that need to be updated by
+ * the Source implementations.
+ */
+public interface SourceMetricGroup extends MetricGroup {
+ // Source metrics convention specified in FLIP-33.
+ String CURRENT_FETCH_EVENT_TIME_LAG = "currentFetchEventTimeLag";
+ String CURRENT_EMIT_EVENT_TIME_LAG = "currentEmitEventTimeLag";
+ String WATERMARK_LAG = "watermarkLag";
+ String SOURCE_IDLE_TIME = "sourceIdleTime";
+ String PENDING_BYTES = "pendingBytes";
+ String PENDING_RECORDS = "pendingRecords";
+ String IO_NUM_RECORDS_IN_ERRORS = "numRecordsInErrors";
+
+ /**
+ * @return the counter to record number of records received by the SourceReader.
+ */
+ Counter getNumRecordsInCounter();
+
+ /**
+ * @return the counter to record number of bytes received by the SourceReader.
+ */
+ Counter getNumBytesInCounter();
+
+ /**
+ * @return the counter to record the errors encountered when reading the records.
+ */
+ Counter getNumRecordsInErrorsCounter();
+
+}
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 518ce86..cc961eb 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
@@ -24,9 +24,10 @@ 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.SourceReaderContext;
+import org.apache.flink.api.connector.source.metrics.SourceMetricGroup;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.InputStatus;
-import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.junit.Test;
@@ -106,8 +107,8 @@ public class NumberSequenceSourceTest {
private static final class DummyReaderContext implements SourceReaderContext {
@Override
- public MetricGroup metricGroup() {
- return new UnregisteredMetricsGroup();
+ public SourceMetricGroup metricGroup() {
+ return new DummySourceMetricsGroup();
}
@Override
@@ -168,4 +169,23 @@ public class NumberSequenceSourceTest {
return emittedRecords;
}
}
+
+ private static final class DummySourceMetricsGroup
+ extends UnregisteredMetricsGroup implements SourceMetricGroup {
+
+ @Override
+ public Counter getNumRecordsInCounter() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Counter getNumBytesInCounter() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Counter getNumRecordsInErrorsCounter() {
+ throw new UnsupportedOperationException();
+ }
+ }
}
diff --git a/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSourceMetricGroup.java b/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSourceMetricGroup.java
new file mode 100644
index 0000000..0b3cae4
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSourceMetricGroup.java
@@ -0,0 +1,50 @@
+/*
+ * 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.api.connector.source.mocks;
+
+import org.apache.flink.api.connector.source.metrics.SourceMetricGroup;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.SimpleCounter;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+
+/**
+ * A testing implementation of {@link org.apache.flink.api.connector.source.metrics.SourceMetricGroup}.
+ * This class is not supposed to be used by classes out of the flink-core module. Please use
+ * TestingSourceMetricGroup in flink-connector-test-utils instead.
+ */
+public class MockSourceMetricGroup extends UnregisteredMetricsGroup implements SourceMetricGroup {
+ private final Counter numRecordsInCounter = new SimpleCounter();
+ private final Counter numBytesInCounter = new SimpleCounter();
+ private final Counter numRecordsInErrorsCounter = new SimpleCounter();
+
+ @Override
+ public Counter getNumRecordsInCounter() {
+ return numRecordsInCounter;
+ }
+
+ @Override
+ public Counter getNumBytesInCounter() {
+ return numBytesInCounter;
+ }
+
+ @Override
+ public Counter getNumRecordsInErrorsCounter() {
+ return numRecordsInErrorsCounter;
+ }
+}
diff --git a/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSourceReader.java b/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSourceReader.java
index 8eb4596..38805ad 100644
--- a/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSourceReader.java
+++ b/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSourceReader.java
@@ -18,6 +18,7 @@
package org.apache.flink.api.connector.source.mocks;
+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.SourceReader;
@@ -43,6 +44,7 @@ public class MockSourceReader implements SourceReader<Integer, MockSourceSplit>
private boolean started;
private boolean closed;
private boolean waitingForMoreSplits;
+ private boolean emitWatermark;
@GuardedBy("this")
private CompletableFuture<Void> availableFuture;
@@ -76,7 +78,11 @@ public class MockSourceReader implements SourceReader<Integer, MockSourceSplit>
}
// Read from the split with available record.
if (currentSplitIndex < assignedSplits.size()) {
- sourceOutput.collect(assignedSplits.get(currentSplitIndex).getNext(false)[0]);
+ int value = assignedSplits.get(currentSplitIndex).getNext(false)[0];
+ sourceOutput.collect(value, value);
+ if (emitWatermark) {
+ sourceOutput.emitWatermark(new Watermark(value));
+ }
return InputStatus.MORE_AVAILABLE;
} else if (finished) {
// In case no split has available record, return depending on whether all the splits has finished.
@@ -153,6 +159,10 @@ public class MockSourceReader implements SourceReader<Integer, MockSourceSplit>
}
}
+ public void enableWatermarkEmission() {
+ emitWatermark = true;
+ }
+
public boolean isStarted() {
return started;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/SourceMetricGroupImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/SourceMetricGroupImpl.java
new file mode 100644
index 0000000..d2b67bf
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/SourceMetricGroupImpl.java
@@ -0,0 +1,76 @@
+/*
+ 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.runtime.metrics.groups;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.connector.source.metrics.SourceMetricGroup;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.util.clock.SystemClock;
+
+import java.util.function.Supplier;
+
+/**
+ * A metric group that adds source specific metrics to the SourceOperator.
+ * It adds Source specific metrics to the OperatorMetricsGroup.
+ */
+public class SourceMetricGroupImpl extends ProxyMetricGroup<OperatorMetricGroup> implements SourceMetricGroup {
+ private final Counter numRecordsInCounter;
+ private final Counter numRecordsInErrorsCounter;
+ private final Counter numBytesIn;
+ private final Supplier<Long> currentWatermarkSupplier;
+
+ public SourceMetricGroupImpl(
+ OperatorMetricGroup operatorMetricGroup,
+ Counter numBytesInCounter,
+ Supplier<Long> currentWatermarkSupplier) {
+ super(operatorMetricGroup);
+ this.numRecordsInCounter = operatorMetricGroup.getIOMetricGroup().getNumRecordsInCounter();
+ this.numRecordsInErrorsCounter = operatorMetricGroup.counter(SourceMetricGroup.IO_NUM_RECORDS_IN_ERRORS);
+ this.numBytesIn = numBytesInCounter;
+ this.currentWatermarkSupplier = currentWatermarkSupplier;
+ operatorMetricGroup.gauge(
+ SourceMetricGroup.WATERMARK_LAG,
+ () -> {
+ long currentWatermark = currentWatermarkSupplier.get();
+ if (currentWatermark < 0) {
+ return -1L;
+ } else {
+ return SystemClock.getInstance().absoluteTimeMillis() - currentWatermark;
+ }
+ });
+ }
+
+ @Override
+ public Counter getNumRecordsInCounter() {
+ return numRecordsInCounter;
+ }
+
+ public Counter getNumBytesInCounter() {
+ return numBytesIn;
+ }
+
+ public Counter getNumRecordsInErrorsCounter() {
+ return numRecordsInErrorsCounter;
+ }
+
+ @VisibleForTesting
+ public Supplier<Long> getCurrentWatermarkSupplier() {
+ return currentWatermarkSupplier;
+ }
+}
diff --git a/flink-streaming-java/pom.xml b/flink-streaming-java/pom.xml
index f0bd66b..94c1c16 100644
--- a/flink-streaming-java/pom.xml
+++ b/flink-streaming-java/pom.xml
@@ -90,6 +90,13 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-test-utils</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
<artifactId>flink-runtime_${scala.binary.version}</artifactId>
<version>${project.version}</version>
<scope>test</scope>
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
index 46839b3..5234805 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
@@ -28,10 +28,12 @@ import org.apache.flink.api.connector.source.SourceEvent;
import org.apache.flink.api.connector.source.SourceReader;
import org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.api.connector.source.metrics.SourceMetricGroup;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.InputStatus;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.metrics.groups.SourceMetricGroupImpl;
import org.apache.flink.runtime.operators.coordination.OperatorEvent;
import org.apache.flink.runtime.operators.coordination.OperatorEventGateway;
import org.apache.flink.runtime.operators.coordination.OperatorEventHandler;
@@ -42,10 +44,13 @@ import org.apache.flink.runtime.source.event.RequestSplitEvent;
import org.apache.flink.runtime.source.event.SourceEventWrapper;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.source.TimestampsAndWatermarks;
import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.function.FunctionWithException;
@@ -119,6 +124,9 @@ public class SourceOperator<OUT, SplitT extends SourceSplit>
* but we currently need to instantiate this lazily, because the metric groups exist only later. */
private TimestampsAndWatermarks<OUT> eventTimeLogic;
+ /** The metric group is initialized lazily at runtime. */
+ private SourceMetricGroup sourceMetricGroup;
+
public SourceOperator(
FunctionWithException<SourceReaderContext, SourceReader<OUT, SplitT>, Exception> readerFactory,
OperatorEventGateway operatorEventGateway,
@@ -128,6 +136,22 @@ public class SourceOperator<OUT, SplitT extends SourceSplit>
Configuration configuration,
String localHostname,
boolean emitProgressiveWatermarks) {
+ // The SourceMetricGorup is set to null here because it is set lazily in the setup().
+ this(readerFactory, operatorEventGateway, splitSerializer, watermarkStrategy,
+ timeService, configuration, localHostname, emitProgressiveWatermarks, null);
+ }
+
+ @VisibleForTesting
+ protected SourceOperator(
+ FunctionWithException<SourceReaderContext, SourceReader<OUT, SplitT>, Exception> readerFactory,
+ OperatorEventGateway operatorEventGateway,
+ SimpleVersionedSerializer<SplitT> splitSerializer,
+ WatermarkStrategy<OUT> watermarkStrategy,
+ ProcessingTimeService timeService,
+ Configuration configuration,
+ String localHostname,
+ boolean emitProgressiveWatermarks,
+ SourceMetricGroup sourceMetricGroup) {
this.readerFactory = checkNotNull(readerFactory);
this.operatorEventGateway = checkNotNull(operatorEventGateway);
@@ -137,16 +161,35 @@ public class SourceOperator<OUT, SplitT extends SourceSplit>
this.configuration = checkNotNull(configuration);
this.localHostname = checkNotNull(localHostname);
this.emitProgressiveWatermarks = emitProgressiveWatermarks;
+ this.sourceMetricGroup = sourceMetricGroup;
+ }
+
+ @Override
+ public void setup(
+ StreamTask<?, ?> containingTask,
+ StreamConfig config,
+ Output<StreamRecord<OUT>> output) {
+ super.setup(containingTask, config, output);
+ // Reuse the numBytesInCounter from TaskMetricGroup.
+ sourceMetricGroup = new SourceMetricGroupImpl(
+ super.metrics,
+ super.metrics.parent().getIOMetricGroup().getNumBytesInCounter(),
+ () -> {
+ if (eventTimeLogic != null) {
+ return eventTimeLogic.getWatermark();
+ } else {
+ return -1L;
+ }
+ });
}
@Override
public void open() throws Exception {
- final MetricGroup metricGroup = getMetricGroup();
final SourceReaderContext context = new SourceReaderContext() {
@Override
- public MetricGroup metricGroup() {
- return metricGroup;
+ public SourceMetricGroup metricGroup() {
+ return sourceMetricGroup;
}
@Override
@@ -180,13 +223,13 @@ public class SourceOperator<OUT, SplitT extends SourceSplit>
if (emitProgressiveWatermarks) {
eventTimeLogic = TimestampsAndWatermarks.createProgressiveEventTimeLogic(
watermarkStrategy,
- metricGroup,
+ sourceMetricGroup,
getProcessingTimeService(),
getExecutionConfig().getAutoWatermarkInterval());
} else {
eventTimeLogic = TimestampsAndWatermarks.createNoOpEventTimeLogic(
watermarkStrategy,
- metricGroup);
+ sourceMetricGroup);
}
sourceReader = readerFactory.apply(context);
@@ -218,6 +261,11 @@ public class SourceOperator<OUT, SplitT extends SourceSplit>
}
@Override
+ public MetricGroup getMetricGroup() {
+ return sourceMetricGroup;
+ }
+
+ @Override
public InputStatus emitNext(DataOutput<OUT> output) throws Exception {
// guarding an assumptions we currently make due to the fact that certain classes
// assume a constant output
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/NoOpTimestampsAndWatermarks.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/NoOpTimestampsAndWatermarks.java
index f28ef55..d263bd6 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/NoOpTimestampsAndWatermarks.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/NoOpTimestampsAndWatermarks.java
@@ -64,6 +64,12 @@ public class NoOpTimestampsAndWatermarks<T> implements TimestampsAndWatermarks<T
// no periodic watermarks
}
+ @Override
+ public long getWatermark() {
+ // No watermark.
+ return -1L;
+ }
+
// ------------------------------------------------------------------------
/**
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java
index 1d67af2..5a11a01 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java
@@ -65,6 +65,9 @@ public class ProgressiveTimestampsAndWatermarks<T> implements TimestampsAndWater
private StreamingReaderOutput<T> currentMainOutput;
@Nullable
+ private WatermarkToDataOutput watermarkOutput;
+
+ @Nullable
private ScheduledFuture<?> periodicEmitHandle;
public ProgressiveTimestampsAndWatermarks(
@@ -98,7 +101,7 @@ public class ProgressiveTimestampsAndWatermarks<T> implements TimestampsAndWater
// support re-assigning the underlying output
checkState(currentMainOutput == null && currentPerSplitOutputs == null, "already created a main output");
- final WatermarkOutput watermarkOutput = new WatermarkToDataOutput(output);
+ watermarkOutput = new WatermarkToDataOutput(output);
final WatermarkGenerator<T> watermarkGenerator = watermarksFactory.createWatermarkGenerator(watermarksContext);
currentPerSplitOutputs = new SplitLocalOutputs<>(
@@ -141,6 +144,15 @@ public class ProgressiveTimestampsAndWatermarks<T> implements TimestampsAndWater
}
}
+ @Override
+ public long getWatermark() {
+ if (watermarkOutput != null) {
+ return watermarkOutput.getMaxWatermarkSoFar();
+ } else {
+ return -1L;
+ }
+ }
+
void triggerPeriodicEmit(@SuppressWarnings("unused") long wallClockTimestamp) {
if (currentPerSplitOutputs != null) {
currentPerSplitOutputs.emitPeriodicWatermark();
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java
index 51068d2..d8320e8 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java
@@ -63,6 +63,13 @@ public interface TimestampsAndWatermarks<T> {
*/
void stopPeriodicWatermarkEmits();
+ /**
+ * Get the current watermark.
+ *
+ * @return the current watermark.
+ */
+ long getWatermark();
+
// ------------------------------------------------------------------------
// factories
// ------------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutput.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutput.java
index 3e3b255..8c26d7b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutput.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutput.java
@@ -83,4 +83,8 @@ public final class WatermarkToDataOutput implements WatermarkOutput {
throw new ExceptionInChainedOperatorException(e);
}
}
+
+ public long getMaxWatermarkSoFar() {
+ return maxWatermarkSoFar;
+ }
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java
index 78348d8..541d28c 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java
@@ -28,6 +28,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.io.SimpleVersionedSerialization;
import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.metrics.groups.SourceMetricGroupImpl;
import org.apache.flink.runtime.operators.coordination.MockOperatorEventGateway;
import org.apache.flink.runtime.operators.coordination.OperatorEvent;
import org.apache.flink.runtime.operators.testutils.MockEnvironment;
@@ -42,6 +43,7 @@ import org.apache.flink.runtime.state.StateInitializationContextImpl;
import org.apache.flink.runtime.state.StateSnapshotContextSynchronousImpl;
import org.apache.flink.runtime.state.TestTaskStateManager;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
+import org.apache.flink.streaming.api.operators.source.CollectingDataOutput;
import org.apache.flink.streaming.api.operators.source.TestingSourceOperator;
import org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask;
import org.apache.flink.streaming.runtime.tasks.StreamMockEnvironment;
@@ -179,6 +181,31 @@ public class SourceOperatorTest {
assertEquals(100L, (long) mockSourceReader.getAbortedCheckpoints().get(0));
}
+ @Test
+ public void testMetrics() throws Exception {
+ StateInitializationContext stateContext = getStateContext();
+ operator.initializeState(stateContext);
+ operator.open();
+ mockSourceReader.enableWatermarkEmission();
+ try {
+ MockSourceSplit newSplit = new MockSourceSplit(2);
+ newSplit.addRecord(100);
+ operator.handleOperatorEvent(new AddSplitEvent<>(
+ Collections.singletonList(newSplit), new MockSourceSplitSerializer()));
+ CollectingDataOutput<Integer> output = new CollectingDataOutput<>();
+ SourceMetricGroupImpl metrics =
+ (SourceMetricGroupImpl) operator.getMetricGroup();
+ assertEquals(-1L, (long) metrics.getCurrentWatermarkSupplier().get());
+ operator.emitNext(output);
+ assertEquals("There should be a record and a watermark emission.",
+ 2, output.events.size());
+ assertEquals(100L, (long) metrics.getCurrentWatermarkSupplier().get());
+ }
+ finally {
+ operator.close();
+ }
+ }
+
// ---------------- helper methods -------------------------
private StateInitializationContext getStateContext() throws Exception {
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/source/CollectingDataOutput.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/source/CollectingDataOutput.java
index a536553..a1a3b20 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/source/CollectingDataOutput.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/source/CollectingDataOutput.java
@@ -30,9 +30,9 @@ import java.util.List;
/**
* A test utility implementation of {@link PushingAsyncDataInput.DataOutput} that collects all events.
*/
-final class CollectingDataOutput<E> implements PushingAsyncDataInput.DataOutput<E> {
+public final class CollectingDataOutput<E> implements PushingAsyncDataInput.DataOutput<E> {
- final List<Object> events = new ArrayList<>();
+ public final List<Object> events = new ArrayList<>();
@Override
public void emitWatermark(Watermark watermark) throws Exception {
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/source/TestingSourceOperator.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/source/TestingSourceOperator.java
index 46b1382..61d19b9 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/source/TestingSourceOperator.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/source/TestingSourceOperator.java
@@ -24,6 +24,7 @@ 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.MockSourceSplitSerializer;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.testutils.source.reader.TestingSourceMetricGroup;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.operators.coordination.MockOperatorEventGateway;
import org.apache.flink.runtime.operators.coordination.OperatorEventGateway;
@@ -36,7 +37,7 @@ import org.apache.flink.streaming.util.MockStreamingRuntimeContext;
/**
* A SourceOperator extension to simplify test setup.
*/
-public class TestingSourceOperator<T> extends SourceOperator<T, MockSourceSplit> {
+public class TestingSourceOperator<T> extends SourceOperator<T, MockSourceSplit> {
private static final long serialVersionUID = 1L;
@@ -80,7 +81,8 @@ public class TestingSourceOperator<T> extends SourceOperator<T, MockSourceSplit
timeService,
new Configuration(),
"localhost",
- emitProgressiveWatermarks);
+ emitProgressiveWatermarks,
+ new TestingSourceMetricGroup());
this.subtaskIndex = subtaskIndex;
this.parallelism = parallelism;
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskChainedSourcesCheckpointingTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskChainedSourcesCheckpointingTest.java
index 5f912ab..9e104d8 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskChainedSourcesCheckpointingTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskChainedSourcesCheckpointingTest.java
@@ -137,9 +137,9 @@ public class MultipleInputStreamTaskChainedSourcesCheckpointingTest {
Future<Boolean> checkpointFuture = testHarness.getStreamTask().triggerCheckpointAsync(metaData, barrier.getCheckpointOptions(), false);
processSingleStepUntil(testHarness, checkpointFuture::isDone);
- expectedOutput.add(new StreamRecord<>("42", TimestampAssigner.NO_TIMESTAMP));
- expectedOutput.add(new StreamRecord<>("42", TimestampAssigner.NO_TIMESTAMP));
- expectedOutput.add(new StreamRecord<>("42", TimestampAssigner.NO_TIMESTAMP));
+ expectedOutput.add(new StreamRecord<>("42", 42L));
+ expectedOutput.add(new StreamRecord<>("42", 42L));
+ expectedOutput.add(new StreamRecord<>("42", 42L));
expectedOutput.add(new StreamRecord<>("44", TimestampAssigner.NO_TIMESTAMP));
expectedOutput.add(new StreamRecord<>("44", TimestampAssigner.NO_TIMESTAMP));
expectedOutput.add(new StreamRecord<>("47.0", TimestampAssigner.NO_TIMESTAMP));
@@ -202,7 +202,7 @@ public class MultipleInputStreamTaskChainedSourcesCheckpointingTest {
addSourceRecords(testHarness, 0, 42, 43, 44);
processSingleStepUntil(testHarness, () -> !testHarness.getOutput().isEmpty());
- expectedOutput.add(new StreamRecord<>("42", TimestampAssigner.NO_TIMESTAMP));
+ expectedOutput.add(new StreamRecord<>("42", 42L));
CheckpointBarrier barrier = createBarrier(testHarness);
Future<Boolean> checkpointFuture = testHarness.getStreamTask().triggerCheckpointAsync(metaData, barrier.getCheckpointOptions(), false);
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
index f3b963a..9259750 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
@@ -19,7 +19,6 @@
package org.apache.flink.streaming.runtime.tasks;
import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
@@ -113,8 +112,8 @@ public class MultipleInputStreamTaskTest {
ArrayDeque<Object> expectedOutput = new ArrayDeque<>();
addSourceRecords(testHarness, 1, 42, 43);
- expectedOutput.add(new StreamRecord<>("42", TimestampAssigner.NO_TIMESTAMP));
- expectedOutput.add(new StreamRecord<>("43", TimestampAssigner.NO_TIMESTAMP));
+ expectedOutput.add(new StreamRecord<>("42", 42));
+ expectedOutput.add(new StreamRecord<>("43", 43));
testHarness.processElement(new StreamRecord<>("Hello", initialTime + 1), 0);
expectedOutput.add(new StreamRecord<>("Hello", initialTime + 1));
testHarness.processElement(new StreamRecord<>(42.44d, initialTime + 3), 1);
@@ -419,7 +418,7 @@ public class MultipleInputStreamTaskTest {
testHarness.processElement(new Watermark(initialTime), 0, 1);
addSourceRecords(testHarness, 1, initialTime);
- expectedOutput.add(new StreamRecord<>("" + (initialTime), TimestampAssigner.NO_TIMESTAMP));
+ expectedOutput.add(new StreamRecord<>("" + (initialTime), initialTime));
testHarness.processElement(new Watermark(initialTime), 1, 0);
@@ -443,7 +442,7 @@ public class MultipleInputStreamTaskTest {
testHarness.processElement(new Watermark(initialTime + 3), 0, 1);
addSourceRecords(testHarness, 1, initialTime + 3);
- expectedOutput.add(new StreamRecord<>("" + (initialTime + 3), TimestampAssigner.NO_TIMESTAMP));
+ expectedOutput.add(new StreamRecord<>("" + (initialTime + 3), initialTime + 3));
testHarness.processElement(new Watermark(initialTime + 3), 1, 0);
testHarness.processElement(new Watermark(initialTime + 2), 1, 1);
@@ -463,7 +462,7 @@ public class MultipleInputStreamTaskTest {
testHarness.processElement(new Watermark(initialTime + 4), 0, 1);
addSourceRecords(testHarness, 1, initialTime + 4);
- expectedOutput.add(new StreamRecord<>("" + (initialTime + 4), TimestampAssigner.NO_TIMESTAMP));
+ expectedOutput.add(new StreamRecord<>("" + (initialTime + 4), initialTime + 4));
testHarness.processElement(new Watermark(initialTime + 4), 1, 0);
expectedOutput.add(new Watermark(initialTime + 4));
@@ -518,7 +517,7 @@ public class MultipleInputStreamTaskTest {
// we wake up the source and emit watermark
addSourceRecords(testHarness, 1, initialTime + 5);
testHarness.processAll();
- expectedOutput.add(new StreamRecord<>("" + (initialTime + 5), TimestampAssigner.NO_TIMESTAMP));
+ expectedOutput.add(new StreamRecord<>("" + (initialTime + 5), initialTime + 5));
expectedOutput.add(new Watermark(initialTime + 5));
// the source should go back to being idle immediately, but AbstractStreamOperatorV2
// should have updated it's watermark by then.
@@ -534,7 +533,7 @@ public class MultipleInputStreamTaskTest {
// make source active once again, emit a watermark and go idle again.
addSourceRecords(testHarness, 1, initialTime + 10);
- expectedOutput.add(new StreamRecord<>("" + (initialTime + 10), TimestampAssigner.NO_TIMESTAMP));
+ expectedOutput.add(new StreamRecord<>("" + (initialTime + 10), initialTime + 10));
expectedOutput.add(StreamStatus.ACTIVE);
expectedOutput.add(StreamStatus.IDLE);
testHarness.processAll();
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java
index ca2ac10..5be45d4 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceOperatorStreamTaskTest.java
@@ -18,7 +18,6 @@
package org.apache.flink.streaming.runtime.tasks;
-import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.connector.source.Boundedness;
@@ -143,7 +142,7 @@ public class SourceOperatorStreamTaskTest {
// Build expected output to verify the results
Queue<Object> expectedOutput = new LinkedList<>();
- expectedRecords.forEach(r -> expectedOutput.offer(new StreamRecord<>(r, TimestampAssigner.NO_TIMESTAMP)));
+ expectedRecords.forEach(r -> expectedOutput.offer(new StreamRecord<>(r, r)));
// Add barrier to the expected output.
expectedOutput.add(new CheckpointBarrier(checkpointId, checkpointId, checkpointOptions));
diff --git a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderContext.java b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderContext.java
index 1fda64a..50df6a3 100644
--- a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderContext.java
+++ b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderContext.java
@@ -20,9 +20,8 @@ package org.apache.flink.connector.testutils.source.reader;
import org.apache.flink.api.connector.source.SourceEvent;
import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.metrics.SourceMetricGroup;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import java.util.ArrayList;
import java.util.List;
@@ -32,7 +31,7 @@ import java.util.List;
*/
public class TestingReaderContext implements SourceReaderContext {
- private final UnregisteredMetricsGroup metrics = new UnregisteredMetricsGroup();
+ private final TestingSourceMetricGroup metrics = new TestingSourceMetricGroup();
private final Configuration config;
@@ -51,7 +50,7 @@ public class TestingReaderContext implements SourceReaderContext {
// ------------------------------------------------------------------------
@Override
- public MetricGroup metricGroup() {
+ public SourceMetricGroup metricGroup() {
return metrics;
}
diff --git a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingSourceMetricGroup.java b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingSourceMetricGroup.java
new file mode 100644
index 0000000..3dc33c2
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingSourceMetricGroup.java
@@ -0,0 +1,48 @@
+/*
+ * 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.testutils.source.reader;
+
+import org.apache.flink.api.connector.source.metrics.SourceMetricGroup;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.SimpleCounter;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+
+/**
+ * A testing implementation of {@link SourceMetricGroup}.
+ */
+public class TestingSourceMetricGroup extends UnregisteredMetricsGroup implements SourceMetricGroup {
+ private final Counter numRecordsInCounter = new SimpleCounter();
+ private final Counter numBytesInCounter = new SimpleCounter();
+ private final Counter numRecordsInErrorsCounter = new SimpleCounter();
+
+ @Override
+ public Counter getNumRecordsInCounter() {
+ return numRecordsInCounter;
+ }
+
+ @Override
+ public Counter getNumBytesInCounter() {
+ return numBytesInCounter;
+ }
+
+ @Override
+ public Counter getNumRecordsInErrorsCounter() {
+ return numRecordsInErrorsCounter;
+ }
+}