You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2017/09/21 11:47:00 UTC
[1/2] flink git commit: [FLINK-7552] Extend SinkFunction interface
with SinkContext
Repository: flink
Updated Branches:
refs/heads/master 9b0ba7ba3 -> 6886f638d
[FLINK-7552] Extend SinkFunction interface with SinkContext
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e7996b0d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e7996b0d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e7996b0d
Branch: refs/heads/master
Commit: e7996b0d0ff5fe705a3830f3855f977cad4f0c44
Parents: 9b0ba7b
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Tue Aug 29 15:50:56 2017 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Thu Sep 21 13:46:17 2017 +0200
----------------------------------------------------------------------
.../cassandra/CassandraConnectorITCase.java | 3 +-
.../kafka/FlinkKafkaProducerBaseTest.java | 3 +-
.../connectors/rabbitmq/common/RMQSinkTest.java | 7 +-
.../api/functions/sink/RichSinkFunction.java | 3 -
.../api/functions/sink/SinkContextUtil.java | 56 +++++++++
.../api/functions/sink/SinkFunction.java | 48 +++++++-
.../streaming/api/operators/StreamSink.java | 60 +++++++++-
.../api/functions/PrintSinkFunctionTest.java | 13 ++-
.../functions/sink/SocketClientSinkTest.java | 8 +-
.../api/operators/StreamSinkOperatorTest.java | 117 +++++++++++++++++++
pom.xml | 2 +
11 files changed, 300 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e7996b0d/flink-connectors/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java b/flink-connectors/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java
index bc5d1a8..f52a42c 100644
--- a/flink-connectors/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java
+++ b/flink-connectors/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraConnectorITCase.java
@@ -36,6 +36,7 @@ import org.apache.flink.core.io.InputSplit;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkContextUtil;
import org.apache.flink.streaming.runtime.operators.WriteAheadSinkTestBase;
import com.datastax.driver.core.Cluster;
@@ -459,7 +460,7 @@ public class CassandraConnectorITCase extends WriteAheadSinkTestBase<Tuple3<Stri
sink.open(new Configuration());
for (scala.Tuple3<String, Integer, Integer> value : scalaTupleCollection) {
- sink.invoke(value);
+ sink.invoke(value, SinkContextUtil.forTimestamp(0));
}
sink.close();
http://git-wip-us.apache.org/repos/asf/flink/blob/e7996b0d/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBaseTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBaseTest.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBaseTest.java
index 08c5f01..6b4b6ff 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBaseTest.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBaseTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.testutils.CheckedThread;
import org.apache.flink.core.testutils.MultiShotLatch;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.functions.sink.SinkContextUtil;
import org.apache.flink.streaming.api.operators.StreamSink;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.connectors.kafka.testutils.FakeStandardProducerConfig;
@@ -117,7 +118,7 @@ public class FlinkKafkaProducerBaseTest {
producer.open(new Configuration());
verify(mockPartitioner, times(1)).open(0, 1);
- producer.invoke("foobar");
+ producer.invoke("foobar", SinkContextUtil.forTimestamp(0));
verify(mockPartitioner, times(1)).partition(
"foobar", null, "foobar".getBytes(), DummyFlinkKafkaProducer.DUMMY_TOPIC, new int[] {0, 1, 2, 3});
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e7996b0d/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQSinkTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQSinkTest.java b/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQSinkTest.java
index 540a7ba..4fb6097 100644
--- a/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQSinkTest.java
+++ b/flink-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/common/RMQSinkTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.connectors.rabbitmq.common;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.sink.SinkContextUtil;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSink;
import org.apache.flink.streaming.util.serialization.SerializationSchema;
@@ -91,7 +92,7 @@ public class RMQSinkTest {
public void invokePublishBytesToQueue() throws Exception {
RMQSink<String> rmqSink = createRMQSink();
- rmqSink.invoke(MESSAGE_STR);
+ rmqSink.invoke(MESSAGE_STR, SinkContextUtil.forTimestamp(0));
verify(serializationSchema).serialize(MESSAGE_STR);
verify(channel).basicPublish("", QUEUE_NAME, null, MESSAGE);
}
@@ -101,7 +102,7 @@ public class RMQSinkTest {
RMQSink<String> rmqSink = createRMQSink();
doThrow(IOException.class).when(channel).basicPublish("", QUEUE_NAME, null, MESSAGE);
- rmqSink.invoke("msg");
+ rmqSink.invoke("msg", SinkContextUtil.forTimestamp(0));
}
@Test
@@ -110,7 +111,7 @@ public class RMQSinkTest {
rmqSink.setLogFailuresOnly(true);
doThrow(IOException.class).when(channel).basicPublish("", QUEUE_NAME, null, MESSAGE);
- rmqSink.invoke("msg");
+ rmqSink.invoke("msg", SinkContextUtil.forTimestamp(0));
}
@Test
http://git-wip-us.apache.org/repos/asf/flink/blob/e7996b0d/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/RichSinkFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/RichSinkFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/RichSinkFunction.java
index 64c38b9..66a0c93 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/RichSinkFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/RichSinkFunction.java
@@ -27,7 +27,4 @@ import org.apache.flink.api.common.functions.AbstractRichFunction;
public abstract class RichSinkFunction<IN> extends AbstractRichFunction implements SinkFunction<IN> {
private static final long serialVersionUID = 1L;
-
- public abstract void invoke(IN value) throws Exception;
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e7996b0d/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkContextUtil.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkContextUtil.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkContextUtil.java
new file mode 100644
index 0000000..2749560
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkContextUtil.java
@@ -0,0 +1,56 @@
+/*
+ * 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.streaming.api.functions.sink;
+
+import org.apache.flink.annotation.Internal;
+
+/**
+ * Utility for creating Sink {@link SinkFunction.Context Contexts}.
+ */
+@Internal
+public class SinkContextUtil {
+
+ /**
+ * Creates a {@link SinkFunction.Context} that
+ * throws an exception when trying to access the current watermark or processing time.
+ */
+ public static <T> SinkFunction.Context<T> forTimestamp(long timestamp) {
+ return new SinkFunction.Context<T>() {
+ @Override
+ public long currentProcessingTime() {
+ throw new RuntimeException("Not implemented");
+ }
+
+ @Override
+ public long currentWatermark() {
+ throw new RuntimeException("Not implemented");
+ }
+
+ @Override
+ public long timestamp() {
+ return timestamp;
+ }
+
+ @Override
+ public boolean hasTimestamp() {
+ return true;
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/e7996b0d/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkFunction.java
index cb9e11d..15a77c4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkFunction.java
@@ -35,6 +35,52 @@ public interface SinkFunction<IN> extends Function, Serializable {
*
* @param value The input record.
* @throws Exception
+ * @deprecated Use {@link #invoke(Object, Context)}.
*/
- void invoke(IN value) throws Exception;
+ @Deprecated
+ default void invoke(IN value) throws Exception {
+ }
+
+ /**
+ * Writes the given value to the sink. This function is called for every record.
+ *
+ * @param value The input record.
+ * @param context Additional context about the input record.
+ * @throws Exception
+ */
+ default void invoke(IN value, Context context) throws Exception {
+ invoke(value);
+ }
+
+ /**
+ * Context that {@link SinkFunction SinkFunctions } can use for getting additional data about
+ * an input record.
+ *
+ * <p>The context is only valid for the duration of a
+ * {@link SinkFunction#invoke(Object, Context)} call. Do not store the context and use
+ * afterwards!
+ *
+ * @param <T> The type of elements accepted by the sink.
+ */
+ @Public // Interface might be extended in the future with additional methods.
+ interface Context<T> {
+
+ /** Returns the current processing time. */
+ long currentProcessingTime();
+
+ /** Returns the current event-time watermark. */
+ long currentWatermark();
+
+ /**
+ * Returns the timestamp of the current input record.
+ */
+ long timestamp();
+
+ /**
+ * Checks whether this record has a timestamp.
+ *
+ * @return True if the record has a timestamp, false if not.
+ */
+ boolean hasTimestamp();
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e7996b0d/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java
index d92d789..f4b09af 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java
@@ -19,8 +19,10 @@ package org.apache.flink.streaming.api.operators;
import org.apache.flink.annotation.Internal;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
/**
* A {@link StreamOperator} for executing {@link SinkFunction SinkFunctions}.
@@ -31,14 +33,27 @@ public class StreamSink<IN> extends AbstractUdfStreamOperator<Object, SinkFuncti
private static final long serialVersionUID = 1L;
+ private transient SimpleContext sinkContext;
+
+ /** We listen to this ourselves because we don't have an {@link InternalTimerService}. */
+ private long currentWatermark = Long.MIN_VALUE;
+
public StreamSink(SinkFunction<IN> sinkFunction) {
super(sinkFunction);
chainingStrategy = ChainingStrategy.ALWAYS;
}
@Override
+ public void open() throws Exception {
+ super.open();
+
+ this.sinkContext = new SimpleContext<>(getProcessingTimeService());
+ }
+
+ @Override
public void processElement(StreamRecord<IN> element) throws Exception {
- userFunction.invoke(element.getValue());
+ sinkContext.element = element;
+ userFunction.invoke(element.getValue(), sinkContext);
}
@Override
@@ -48,4 +63,47 @@ public class StreamSink<IN> extends AbstractUdfStreamOperator<Object, SinkFuncti
// sinks don't forward latency markers
}
+
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
+ super.processWatermark(mark);
+ this.currentWatermark = mark.getTimestamp();
+ }
+
+ private class SimpleContext<IN> implements SinkFunction.Context<IN> {
+
+ private StreamRecord<IN> element;
+
+ private final ProcessingTimeService processingTimeService;
+
+ public SimpleContext(ProcessingTimeService processingTimeService) {
+ this.processingTimeService = processingTimeService;
+ }
+
+ @Override
+ public long currentProcessingTime() {
+ return processingTimeService.getCurrentProcessingTime();
+ }
+
+ @Override
+ public long currentWatermark() {
+ return currentWatermark;
+ }
+
+ @Override
+ public long timestamp() {
+ if (!element.hasTimestamp()) {
+ throw new IllegalStateException(
+ "Record has no timestamp. Is the time characteristic set to 'ProcessingTime', or " +
+ "did you forget to call 'DataStream.assignTimestampsAndWatermarks(...)'?");
+
+ }
+ return element.getTimestamp();
+ }
+
+ public boolean hasTimestamp() {
+ return element.hasTimestamp();
+ }
+
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e7996b0d/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkFunctionTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkFunctionTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkFunctionTest.java
index 9e7ecdd..8c303d1 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkFunctionTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkFunctionTest.java
@@ -19,6 +19,7 @@ package org.apache.flink.streaming.api.functions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
+import org.apache.flink.streaming.api.functions.sink.SinkContextUtil;
import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.junit.After;
@@ -40,7 +41,7 @@ public class PrintSinkFunctionTest {
private String line = System.lineSeparator();
@Test
- public void testPrintSinkStdOut(){
+ public void testPrintSinkStdOut() throws Exception {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
PrintStream stream = new PrintStream(baos);
System.setOut(stream);
@@ -55,7 +56,7 @@ public class PrintSinkFunctionTest {
Assert.fail();
}
printSink.setTargetToStandardOut();
- printSink.invoke("hello world!");
+ printSink.invoke("hello world!", SinkContextUtil.forTimestamp(0));
assertEquals("Print to System.out", printSink.toString());
assertEquals("hello world!" + line, baos.toString());
@@ -65,7 +66,7 @@ public class PrintSinkFunctionTest {
}
@Test
- public void testPrintSinkStdErr(){
+ public void testPrintSinkStdErr() throws Exception {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
PrintStream stream = new PrintStream(baos);
System.setOut(stream);
@@ -80,7 +81,7 @@ public class PrintSinkFunctionTest {
Assert.fail();
}
printSink.setTargetToStandardErr();
- printSink.invoke("hello world!");
+ printSink.invoke("hello world!", SinkContextUtil.forTimestamp(0));
assertEquals("Print to System.err", printSink.toString());
assertEquals("hello world!" + line, baos.toString());
@@ -90,7 +91,7 @@ public class PrintSinkFunctionTest {
}
@Test
- public void testPrintSinkWithPrefix(){
+ public void testPrintSinkWithPrefix() throws Exception {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
PrintStream stream = new PrintStream(baos);
System.setOut(stream);
@@ -107,7 +108,7 @@ public class PrintSinkFunctionTest {
Assert.fail();
}
printSink.setTargetToStandardErr();
- printSink.invoke("hello world!");
+ printSink.invoke("hello world!", SinkContextUtil.forTimestamp(0));
assertEquals("Print to System.err", printSink.toString());
assertEquals("2> hello world!" + line, baos.toString());
http://git-wip-us.apache.org/repos/asf/flink/blob/e7996b0d/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/SocketClientSinkTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/SocketClientSinkTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/SocketClientSinkTest.java
index 63e83d2..6cdce11 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/SocketClientSinkTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/SocketClientSinkTest.java
@@ -74,7 +74,7 @@ public class SocketClientSinkTest extends TestLogger {
try {
SocketClientSink<String> simpleSink = new SocketClientSink<>(host, port, simpleSchema, 0);
simpleSink.open(new Configuration());
- simpleSink.invoke(TEST_MESSAGE + '\n');
+ simpleSink.invoke(TEST_MESSAGE + '\n', SinkContextUtil.forTimestamp(0));
simpleSink.close();
}
catch (Throwable t) {
@@ -117,7 +117,7 @@ public class SocketClientSinkTest extends TestLogger {
public void run() {
try {
// need two messages here: send a fin to cancel the client state:FIN_WAIT_2 while the server is CLOSE_WAIT
- simpleSink.invoke(TEST_MESSAGE + '\n');
+ simpleSink.invoke(TEST_MESSAGE + '\n', SinkContextUtil.forTimestamp(0));
}
catch (Throwable t) {
error.set(t);
@@ -182,7 +182,7 @@ public class SocketClientSinkTest extends TestLogger {
// socket should be closed, so this should trigger a re-try
// need two messages here: send a fin to cancel the client state:FIN_WAIT_2 while the server is CLOSE_WAIT
while (true) { // we have to do this more often as the server side closed is not guaranteed to be noticed immediately
- simpleSink.invoke(TEST_MESSAGE + '\n');
+ simpleSink.invoke(TEST_MESSAGE + '\n', SinkContextUtil.forTimestamp(0));
}
}
catch (IOException e) {
@@ -238,7 +238,7 @@ public class SocketClientSinkTest extends TestLogger {
// Initial payload => this will be received by the server an then the socket will be
// closed.
- sink.invoke("0\n");
+ sink.invoke("0\n", SinkContextUtil.forTimestamp(0));
// Get future an make sure there was no problem. This will rethrow any Exceptions from
// the server.
http://git-wip-us.apache.org/repos/asf/flink/blob/e7996b0d/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamSinkOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamSinkOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamSinkOperatorTest.java
new file mode 100644
index 0000000..500a52a
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamSinkOperatorTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.streaming.api.operators;
+
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.Matchers.contains;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for {@link StreamSink}.
+ */
+public class StreamSinkOperatorTest extends TestLogger {
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ /**
+ * Verify that we can correctly query watermark, processing time and the timestamp from the
+ * context.
+ */
+ @Test
+ public void testTimeQuerying() throws Exception {
+
+ BufferingQueryingSink<String> bufferingSink = new BufferingQueryingSink<>();
+
+ StreamSink<String> operator = new StreamSink<>(bufferingSink);
+
+ OneInputStreamOperatorTestHarness<String, Object> testHarness =
+ new OneInputStreamOperatorTestHarness<>(operator);
+
+ testHarness.setup();
+ testHarness.open();
+
+ testHarness.processWatermark(new Watermark(17));
+ testHarness.setProcessingTime(12);
+ testHarness.processElement(new StreamRecord<>("Hello", 12L));
+
+ testHarness.processWatermark(new Watermark(42));
+ testHarness.setProcessingTime(15);
+ testHarness.processElement(new StreamRecord<>("Ciao", 13L));
+
+ testHarness.processWatermark(new Watermark(42));
+ testHarness.setProcessingTime(15);
+ testHarness.processElement(new StreamRecord<>("Ciao"));
+
+ assertThat(bufferingSink.data.size(), is(3));
+
+ assertThat(bufferingSink.data,
+ contains(
+ new Tuple4<>(17L, 12L, 12L, "Hello"),
+ new Tuple4<>(42L, 15L, 13L, "Ciao"),
+ new Tuple4<>(42L, 15L, null, "Ciao")));
+
+ testHarness.close();
+ }
+
+ private static class BufferingQueryingSink<T> implements SinkFunction<T> {
+
+ // watermark, processing-time, timestamp, event
+ private final List<Tuple4<Long, Long, Long, T>> data;
+
+ public BufferingQueryingSink() {
+ data = new ArrayList<>();
+ }
+
+ @Override
+ public void invoke(
+ T value, Context context) throws Exception {
+ if (context.hasTimestamp()) {
+ data.add(
+ new Tuple4<>(
+ context.currentWatermark(),
+ context.currentProcessingTime(),
+ context.timestamp(),
+ value));
+ } else {
+ data.add(
+ new Tuple4<>(
+ context.currentWatermark(),
+ context.currentProcessingTime(),
+ null,
+ value));
+
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/e7996b0d/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 561ec79..229e93d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1457,6 +1457,8 @@ under the License.
<excludes>
<exclude>@org.apache.flink.annotation.PublicEvolving</exclude>
<exclude>@org.apache.flink.annotation.Internal</exclude>
+ <exclude>org.apache.flink.streaming.api.functions.sink.RichSinkFunction#invoke(java.lang.Object)</exclude>
+ <exclude>org.apache.flink.streaming.api.functions.sink.SinkFunction</exclude>
</excludes>
<accessModifier>public</accessModifier>
<breakBuildOnModifications>false</breakBuildOnModifications>
[2/2] flink git commit: [FLINK-7553] Use new SinkFunction interface
in FlinkKafkaProducer010
Posted by al...@apache.org.
[FLINK-7553] Use new SinkFunction interface in FlinkKafkaProducer010
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6886f638
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6886f638
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6886f638
Branch: refs/heads/master
Commit: 6886f638d70419e01ebdfc1cdbb6d834f3fb30b0
Parents: e7996b0
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Tue Aug 29 15:53:16 2017 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Thu Sep 21 13:46:39 2017 +0200
----------------------------------------------------------------------
.../connectors/kafka/FlinkKafkaProducer010.java | 226 +++++--------------
.../kafka/FlinkKafkaProducerBase.java | 2 +-
.../connectors/kafka/KafkaProducerTestBase.java | 3 +
.../api/functions/sink/SinkContextUtil.java | 7 +-
.../api/functions/sink/SinkFunction.java | 26 +--
.../streaming/api/operators/StreamSink.java | 16 +-
.../api/operators/StreamSinkOperatorTest.java | 3 +-
7 files changed, 81 insertions(+), 202 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6886f638/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java b/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java
index 3b9dff1..3b43a7e 100644
--- a/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java
+++ b/flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java
@@ -17,24 +17,13 @@
package org.apache.flink.streaming.connectors.kafka;
-import org.apache.flink.api.common.functions.IterationRuntimeContext;
-import org.apache.flink.api.common.functions.RichFunction;
-import org.apache.flink.api.common.functions.RuntimeContext;
-import org.apache.flink.api.java.typeutils.GenericTypeInfo;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.state.FunctionInitializationContext;
-import org.apache.flink.runtime.state.FunctionSnapshotContext;
-import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
-import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-import org.apache.flink.streaming.api.operators.StreamSink;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaDelegatePartitioner;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper;
import org.apache.flink.streaming.util.serialization.SerializationSchema;
@@ -43,32 +32,10 @@ import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
-import static org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase.getPartitionsByTopic;
-import static org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase.getPropertiesFromBrokerList;
-
/**
* Flink Sink to produce data into a Kafka topic. This producer is compatible with Kafka 0.10.x
- *
- * <p>Implementation note: This producer is a hybrid between a regular regular sink function (a)
- * and a custom operator (b).
- *
- * <p>For (a), the class implements the SinkFunction and RichFunction interfaces.
- * For (b), it extends the StreamTask class.
- *
- * <p>Details about approach (a):
- * Pre Kafka 0.10 producers only follow approach (a), allowing users to use the producer using the
- * DataStream.addSink() method.
- * Since the APIs exposed in that variant do not allow accessing the the timestamp attached to the record
- * the Kafka 0.10 producer has a second invocation option, approach (b).
- *
- * <p>Details about approach (b):
- * Kafka 0.10 supports writing the timestamp attached to a record to Kafka. When adding the
- * FlinkKafkaProducer010 using the FlinkKafkaProducer010.writeToKafkaWithTimestamps() method, the Kafka producer
- * can access the internal record timestamp of the record and write it to Kafka.
- *
- * <p>All methods and constructors in this class are marked with the approach they are needed for.
*/
-public class FlinkKafkaProducer010<T> extends StreamSink<T> implements SinkFunction<T>, RichFunction, CheckpointedFunction {
+public class FlinkKafkaProducer010<T> extends FlinkKafkaProducer09<T> {
/**
* Flag controlling whether we are writing the Flink record's timestamp into Kafka.
@@ -87,7 +54,11 @@ public class FlinkKafkaProducer010<T> extends StreamSink<T> implements SinkFunct
* @param topicId ID of the Kafka topic.
* @param serializationSchema User defined serialization schema supporting key/value messages
* @param producerConfig Properties with the producer configuration.
+ *
+ * @deprecated Use {@link #FlinkKafkaProducer010(String, KeyedSerializationSchema, Properties)}
+ * and call {@link #setWriteTimestampToKafka(boolean)}.
*/
+ @Deprecated
public static <T> FlinkKafkaProducer010Configuration<T> writeToKafkaWithTimestamps(DataStream<T> inStream,
String topicId,
KeyedSerializationSchema<T> serializationSchema,
@@ -105,7 +76,11 @@ public class FlinkKafkaProducer010<T> extends StreamSink<T> implements SinkFunct
* @param topicId ID of the Kafka topic.
* @param serializationSchema User defined (keyless) serialization schema.
* @param producerConfig Properties with the producer configuration.
+ *
+ * @deprecated Use {@link #FlinkKafkaProducer010(String, KeyedSerializationSchema, Properties)}
+ * and call {@link #setWriteTimestampToKafka(boolean)}.
*/
+ @Deprecated
public static <T> FlinkKafkaProducer010Configuration<T> writeToKafkaWithTimestamps(DataStream<T> inStream,
String topicId,
SerializationSchema<T> serializationSchema,
@@ -124,20 +99,24 @@ public class FlinkKafkaProducer010<T> extends StreamSink<T> implements SinkFunct
* @param serializationSchema A serializable serialization schema for turning user objects into a kafka-consumable byte[] supporting key/value messages
* @param producerConfig Configuration properties for the KafkaProducer. 'bootstrap.servers.' is the only required argument.
* @param customPartitioner A serializable partitioner for assigning messages to Kafka partitions.
+ *
+ * @deprecated Use {@link #FlinkKafkaProducer010(String, KeyedSerializationSchema, Properties, FlinkKafkaPartitioner)}
+ * and call {@link #setWriteTimestampToKafka(boolean)}.
*/
+ @Deprecated
public static <T> FlinkKafkaProducer010Configuration<T> writeToKafkaWithTimestamps(DataStream<T> inStream,
String topicId,
KeyedSerializationSchema<T> serializationSchema,
Properties producerConfig,
FlinkKafkaPartitioner<T> customPartitioner) {
- GenericTypeInfo<Object> objectTypeInfo = new GenericTypeInfo<>(Object.class);
FlinkKafkaProducer010<T> kafkaProducer = new FlinkKafkaProducer010<>(topicId, serializationSchema, producerConfig, customPartitioner);
- SingleOutputStreamOperator<Object> transformation = inStream.transform("FlinKafkaProducer 0.10.x", objectTypeInfo, kafkaProducer);
- return new FlinkKafkaProducer010Configuration<>(transformation, kafkaProducer);
+ DataStreamSink<T> streamSink = inStream.addSink(kafkaProducer);
+ return new FlinkKafkaProducer010Configuration<>(streamSink, inStream, kafkaProducer);
+
}
- // ---------------------- Regular constructors w/o timestamp support ------------------
+ // ---------------------- Regular constructors------------------
/**
* Creates a FlinkKafkaProducer for a given topic. The sink produces a DataStream to
@@ -220,9 +199,7 @@ public class FlinkKafkaProducer010<T> extends StreamSink<T> implements SinkFunct
* <p>This constructor does not allow writing timestamps to Kafka, it follow approach (a) (see above)
*/
public FlinkKafkaProducer010(String topicId, KeyedSerializationSchema<T> serializationSchema, Properties producerConfig, FlinkKafkaPartitioner<T> customPartitioner) {
- // We create a Kafka 09 producer instance here and only "override" (by intercepting) the
- // invoke call.
- super(new FlinkKafkaProducer09<>(topicId, serializationSchema, producerConfig, customPartitioner));
+ super(topicId, serializationSchema, producerConfig, customPartitioner);
}
// ----------------------------- Deprecated constructors / factory methods ---------------------------
@@ -250,11 +227,10 @@ public class FlinkKafkaProducer010<T> extends StreamSink<T> implements SinkFunct
Properties producerConfig,
KafkaPartitioner<T> customPartitioner) {
- GenericTypeInfo<Object> objectTypeInfo = new GenericTypeInfo<>(Object.class);
FlinkKafkaProducer010<T> kafkaProducer =
new FlinkKafkaProducer010<>(topicId, serializationSchema, producerConfig, new FlinkKafkaDelegatePartitioner<>(customPartitioner));
- SingleOutputStreamOperator<Object> transformation = inStream.transform("FlinKafkaProducer 0.10.x", objectTypeInfo, kafkaProducer);
- return new FlinkKafkaProducer010Configuration<>(transformation, kafkaProducer);
+ DataStreamSink<T> streamSink = inStream.addSink(kafkaProducer);
+ return new FlinkKafkaProducer010Configuration<T>(streamSink, inStream, kafkaProducer);
}
/**
@@ -288,157 +264,75 @@ public class FlinkKafkaProducer010<T> extends StreamSink<T> implements SinkFunct
public FlinkKafkaProducer010(String topicId, KeyedSerializationSchema<T> serializationSchema, Properties producerConfig, KafkaPartitioner<T> customPartitioner) {
// We create a Kafka 09 producer instance here and only "override" (by intercepting) the
// invoke call.
- super(new FlinkKafkaProducer09<>(topicId, serializationSchema, producerConfig, new FlinkKafkaDelegatePartitioner<>(customPartitioner)));
+ super(topicId, serializationSchema, producerConfig, customPartitioner);
}
- // ----------------------------- Generic element processing ---------------------------
+ /**
+ * If set to true, Flink will write the (event time) timestamp attached to each record into Kafka.
+ * Timestamps must be positive for Kafka to accept them.
+ *
+ * @param writeTimestampToKafka Flag indicating if Flink's internal timestamps are written to Kafka.
+ */
+ public void setWriteTimestampToKafka(boolean writeTimestampToKafka) {
+ this.writeTimestampToKafka = writeTimestampToKafka;
+ }
- private void invokeInternal(T next, long elementTimestamp) throws Exception {
- final FlinkKafkaProducerBase<T> internalProducer = (FlinkKafkaProducerBase<T>) userFunction;
+ // ----------------------------- Generic element processing ---------------------------
- internalProducer.checkErroneous();
+ @Override
+ public void invoke(T value, Context context) throws Exception {
- byte[] serializedKey = internalProducer.schema.serializeKey(next);
- byte[] serializedValue = internalProducer.schema.serializeValue(next);
- String targetTopic = internalProducer.schema.getTargetTopic(next);
+ checkErroneous();
+
+ byte[] serializedKey = schema.serializeKey(value);
+ byte[] serializedValue = schema.serializeValue(value);
+ String targetTopic = schema.getTargetTopic(value);
if (targetTopic == null) {
- targetTopic = internalProducer.defaultTopicId;
+ targetTopic = defaultTopicId;
}
Long timestamp = null;
if (this.writeTimestampToKafka) {
- timestamp = elementTimestamp;
+ timestamp = context.timestamp();
}
ProducerRecord<byte[], byte[]> record;
- int[] partitions = internalProducer.topicPartitionsMap.get(targetTopic);
+ int[] partitions = topicPartitionsMap.get(targetTopic);
if (null == partitions) {
- partitions = getPartitionsByTopic(targetTopic, internalProducer.producer);
- internalProducer.topicPartitionsMap.put(targetTopic, partitions);
+ partitions = getPartitionsByTopic(targetTopic, producer);
+ topicPartitionsMap.put(targetTopic, partitions);
}
- if (internalProducer.flinkKafkaPartitioner == null) {
+ if (flinkKafkaPartitioner == null) {
record = new ProducerRecord<>(targetTopic, null, timestamp, serializedKey, serializedValue);
} else {
- record = new ProducerRecord<>(targetTopic, internalProducer.flinkKafkaPartitioner.partition(next, serializedKey, serializedValue, targetTopic, partitions), timestamp, serializedKey, serializedValue);
+ record = new ProducerRecord<>(targetTopic, flinkKafkaPartitioner.partition(value, serializedKey, serializedValue, targetTopic, partitions), timestamp, serializedKey, serializedValue);
}
- if (internalProducer.flushOnCheckpoint) {
- synchronized (internalProducer.pendingRecordsLock) {
- internalProducer.pendingRecords++;
+ if (flushOnCheckpoint) {
+ synchronized (pendingRecordsLock) {
+ pendingRecords++;
}
}
- internalProducer.producer.send(record, internalProducer.callback);
- }
-
- // ----------------- Helper methods implementing methods from SinkFunction and RichFunction (Approach (a)) ----
-
- // ---- Configuration setters
-
- /**
- * Defines whether the producer should fail on errors, or only log them.
- * If this is set to true, then exceptions will be only logged, if set to false,
- * exceptions will be eventually thrown and cause the streaming program to
- * fail (and enter recovery).
- *
- * <p>Method is only accessible for approach (a) (see above)
- *
- * @param logFailuresOnly The flag to indicate logging-only on exceptions.
- */
- public void setLogFailuresOnly(boolean logFailuresOnly) {
- final FlinkKafkaProducerBase<T> internalProducer = (FlinkKafkaProducerBase<T>) userFunction;
- internalProducer.setLogFailuresOnly(logFailuresOnly);
- }
-
- /**
- * If set to true, the Flink producer will wait for all outstanding messages in the Kafka buffers
- * to be acknowledged by the Kafka producer on a checkpoint.
- * This way, the producer can guarantee that messages in the Kafka buffers are part of the checkpoint.
- *
- * <p>Method is only accessible for approach (a) (see above)
- *
- * @param flush Flag indicating the flushing mode (true = flush on checkpoint)
- */
- public void setFlushOnCheckpoint(boolean flush) {
- final FlinkKafkaProducerBase<T> internalProducer = (FlinkKafkaProducerBase<T>) userFunction;
- internalProducer.setFlushOnCheckpoint(flush);
- }
-
- /**
- * This method is used for approach (a) (see above).
- */
- @Override
- public void open(Configuration parameters) throws Exception {
- final FlinkKafkaProducerBase<T> internalProducer = (FlinkKafkaProducerBase<T>) userFunction;
- internalProducer.open(parameters);
- }
-
- /**
- * This method is used for approach (a) (see above).
- */
- @Override
- public IterationRuntimeContext getIterationRuntimeContext() {
- final FlinkKafkaProducerBase<T> internalProducer = (FlinkKafkaProducerBase<T>) userFunction;
- return internalProducer.getIterationRuntimeContext();
- }
-
- /**
- * This method is used for approach (a) (see above).
- */
- @Override
- public void setRuntimeContext(RuntimeContext t) {
- final FlinkKafkaProducerBase<T> internalProducer = (FlinkKafkaProducerBase<T>) userFunction;
- internalProducer.setRuntimeContext(t);
- }
-
- /**
- * Invoke method for using the Sink as DataStream.addSink() sink.
- *
- * <p>This method is used for approach (a) (see above)
- *
- * @param value The input record.
- */
- @Override
- public void invoke(T value) throws Exception {
- invokeInternal(value, Long.MAX_VALUE);
- }
-
- // ----------------- Helper methods and classes implementing methods from StreamSink (Approach (b)) ----
-
- /**
- * Process method for using the sink with timestamp support.
- *
- * <p>This method is used for approach (b) (see above)
- */
- @Override
- public void processElement(StreamRecord<T> element) throws Exception {
- invokeInternal(element.getValue(), element.getTimestamp());
- }
-
- @Override
- public void initializeState(FunctionInitializationContext context) throws Exception {
- final FlinkKafkaProducerBase<T> internalProducer = (FlinkKafkaProducerBase<T>) userFunction;
- internalProducer.initializeState(context);
- }
-
- @Override
- public void snapshotState(FunctionSnapshotContext context) throws Exception {
- final FlinkKafkaProducerBase<T> internalProducer = (FlinkKafkaProducerBase<T>) userFunction;
- internalProducer.snapshotState(context);
+ producer.send(record, callback);
}
/**
* Configuration object returned by the writeToKafkaWithTimestamps() call.
+ *
+ * <p>This is only kept because it's part of the public API. It is not necessary anymore, now
+ * that the {@link SinkFunction} interface provides timestamps.
*/
public static class FlinkKafkaProducer010Configuration<T> extends DataStreamSink<T> {
- private final FlinkKafkaProducerBase wrappedProducerBase;
private final FlinkKafkaProducer010 producer;
- private FlinkKafkaProducer010Configuration(DataStream stream, FlinkKafkaProducer010<T> producer) {
+ private FlinkKafkaProducer010Configuration(
+ DataStreamSink originalSink,
+ DataStream<T> inputStream,
+ FlinkKafkaProducer010<T> producer) {
//noinspection unchecked
- super(stream, producer);
+ super(inputStream, originalSink.getTransformation().getOperator());
this.producer = producer;
- this.wrappedProducerBase = (FlinkKafkaProducerBase) producer.userFunction;
}
/**
@@ -450,7 +344,7 @@ public class FlinkKafkaProducer010<T> extends StreamSink<T> implements SinkFunct
* @param logFailuresOnly The flag to indicate logging-only on exceptions.
*/
public void setLogFailuresOnly(boolean logFailuresOnly) {
- this.wrappedProducerBase.setLogFailuresOnly(logFailuresOnly);
+ producer.setLogFailuresOnly(logFailuresOnly);
}
/**
@@ -461,7 +355,7 @@ public class FlinkKafkaProducer010<T> extends StreamSink<T> implements SinkFunct
* @param flush Flag indicating the flushing mode (true = flush on checkpoint)
*/
public void setFlushOnCheckpoint(boolean flush) {
- this.wrappedProducerBase.setFlushOnCheckpoint(flush);
+ producer.setFlushOnCheckpoint(flush);
}
/**
@@ -471,7 +365,7 @@ public class FlinkKafkaProducer010<T> extends StreamSink<T> implements SinkFunct
* @param writeTimestampToKafka Flag indicating if Flink's internal timestamps are written to Kafka.
*/
public void setWriteTimestampToKafka(boolean writeTimestampToKafka) {
- this.producer.writeTimestampToKafka = writeTimestampToKafka;
+ producer.writeTimestampToKafka = writeTimestampToKafka;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6886f638/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBase.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBase.java
index 76a2f84..befc1a1 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBase.java
@@ -276,7 +276,7 @@ public abstract class FlinkKafkaProducerBase<IN> extends RichSinkFunction<IN> im
* The incoming data
*/
@Override
- public void invoke(IN next) throws Exception {
+ public void invoke(IN next, Context context) throws Exception {
// propagate asynchronous errors
checkErroneous();
http://git-wip-us.apache.org/repos/asf/flink/blob/6886f638/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
index 000de52..35607dd 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
@@ -215,6 +215,9 @@ public abstract class KafkaProducerTestBase extends KafkaTestBase {
* This test sets KafkaProducer so that it will not automatically flush the data and
* simulate network failure between Flink and Kafka to check whether FlinkKafkaProducer
* flushed records manually on snapshotState.
+ *
+ * <p>Due to legacy reasons there are two different ways of instantiating a Kafka 0.10 sink. The
+ * parameter controls which method is used.
*/
protected void testOneToOneAtLeastOnce(boolean regularSink) throws Exception {
final String topic = regularSink ? "oneToOneTopicRegularSink" : "oneToOneTopicCustomOperator";
http://git-wip-us.apache.org/repos/asf/flink/blob/6886f638/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkContextUtil.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkContextUtil.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkContextUtil.java
index 2749560..3b02ad0 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkContextUtil.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkContextUtil.java
@@ -43,14 +43,9 @@ public class SinkContextUtil {
}
@Override
- public long timestamp() {
+ public Long timestamp() {
return timestamp;
}
-
- @Override
- public boolean hasTimestamp() {
- return true;
- }
};
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6886f638/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkFunction.java
index 15a77c4..74870bc 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkFunction.java
@@ -31,22 +31,22 @@ import java.io.Serializable;
public interface SinkFunction<IN> extends Function, Serializable {
/**
- * Function for standard sink behaviour. This function is called for every record.
- *
- * @param value The input record.
- * @throws Exception
* @deprecated Use {@link #invoke(Object, Context)}.
*/
@Deprecated
- default void invoke(IN value) throws Exception {
- }
+ default void invoke(IN value) throws Exception {}
/**
* Writes the given value to the sink. This function is called for every record.
*
+ * <p>You have to override this method when implementing a {@code SinkFunction}, this is a
+ * {@code default} method for backward compatibility with the old-style method only.
+ *
* @param value The input record.
* @param context Additional context about the input record.
- * @throws Exception
+ *
+ * @throws Exception This method may throw exceptions. Throwing an exception will cause the operation
+ * to fail and may trigger recovery.
*/
default void invoke(IN value, Context context) throws Exception {
invoke(value);
@@ -72,15 +72,9 @@ public interface SinkFunction<IN> extends Function, Serializable {
long currentWatermark();
/**
- * Returns the timestamp of the current input record.
- */
- long timestamp();
-
- /**
- * Checks whether this record has a timestamp.
- *
- * @return True if the record has a timestamp, false if not.
+ * Returns the timestamp of the current input record or {@code null} if the element does not
+ * have an assigned timestamp.
*/
- boolean hasTimestamp();
+ Long timestamp();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6886f638/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java
index f4b09af..667e130 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java
@@ -91,19 +91,11 @@ public class StreamSink<IN> extends AbstractUdfStreamOperator<Object, SinkFuncti
}
@Override
- public long timestamp() {
- if (!element.hasTimestamp()) {
- throw new IllegalStateException(
- "Record has no timestamp. Is the time characteristic set to 'ProcessingTime', or " +
- "did you forget to call 'DataStream.assignTimestampsAndWatermarks(...)'?");
-
+ public Long timestamp() {
+ if (element.hasTimestamp()) {
+ return element.getTimestamp();
}
- return element.getTimestamp();
- }
-
- public boolean hasTimestamp() {
- return element.hasTimestamp();
+ return null;
}
-
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6886f638/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamSinkOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamSinkOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamSinkOperatorTest.java
index 500a52a..9085ade 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamSinkOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamSinkOperatorTest.java
@@ -96,7 +96,8 @@ public class StreamSinkOperatorTest extends TestLogger {
@Override
public void invoke(
T value, Context context) throws Exception {
- if (context.hasTimestamp()) {
+ Long timestamp = context.timestamp();
+ if (timestamp != null) {
data.add(
new Tuple4<>(
context.currentWatermark(),