Github user EronWright commented on a diff in the pull request:
https://github.com/apache/flink/pull/4616#discussion_r137082133
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SinkFunction.java
---
@@ -35,6 +35,39 @@
*
* @param value The input record.
* @throws Exception
+ * @deprecated Use {@link #invoke(SinkContext, Object)}.
*/
- 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 context Additional context about the input record.
+ * @param value The input record.
+ * @throws Exception
+ */
+ default void invoke(SinkContext context, IN value) 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(SinkContext, Object)} 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 SinkContext<T> {
+
+ /**
+ * Returns the timestamp of the current input record.
+ */
+ long timestamp();
--- End diff --
@pnowojski let us not throw an exception here, given the commented-out
section of `StreamRecord::getTimestamp` which suggests a problem with that
approach. It would be good to know why `ProcessFunction` used a `Long` rather
than `long`. I have an almost unhealthy desire for consistency.
---
