[FLINK-7553] Use new SinkFunction interface in FlinkKafkaProducer010

flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBase.java

@@ -276,7 +276,7 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
 	 * 		The incoming data
 	 */
 	@Override
-	public void invoke(IN next) throws Exception {
+	public void invoke(IN next, Context context) throws Exception {
 		// propagate asynchronous errors
 		checkErroneous();
 

flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java

@@ -215,6 +215,9 @@ public void testOneToOneAtLeastOnceCustomOperator() throws Exception {
 	 * 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";

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkContextUtil.java

@@ -43,14 +43,9 @@ public long currentWatermark() {
 			}
 
 			@Override
-			public long timestamp() {
+			public Long timestamp() {
 				return timestamp;
 			}
-
-			@Override
-			public boolean hasTimestamp() {
-				return true;
-			}
 		};
 	}
 }

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkFunction.java

@@ -31,22 +31,22 @@
 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 @@ interface Context<T> {
 		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();
 	}
 }

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSink.java

@@ -91,19 +91,11 @@ public long 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(...)'?");
-
+		public Long timestamp() {
+			if (element.hasTimestamp()) {
+				return element.getTimestamp();
 			}
-			return element.getTimestamp();
-		}
-
-		public boolean hasTimestamp() {
-			return element.hasTimestamp();
+			return null;
 		}
-
 	}
 }

flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamSinkOperatorTest.java

@@ -96,7 +96,8 @@ public BufferingQueryingSink() {
 		@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(),