Status

...

...

...

...

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

...

package org.apache.flink.api.connector.sink;
import org.apache.flink.api.common.eventtime.Watermark;

@Experimental
public interface SinkWriter<InputT, CommT, WriterStateT> extends AutoCloseable {

  /**
   * Add a watermark to the writer.
   *
   * @param watermark The watermark.
   * @throws IOException if fail to add a watermark.
   */
  default void writeWatermark(Watermark watermark) throws IOException {}

  void write(InputT element, Context context) throws IOException;
}

...

Compatibility, Deprecation, and Migration Plan

No compatibility issues are anticipated.   SpecificallyRegarding backwards compatibility:

1. The interface methods have a default implementation.
2. The proposed method name (writeWatermark) is new to the Flink APIs to further minimize the impact. avoid possible collisions.

Regarding binary compatibility of API signatures (japicmp):

1. SinkFunction is already excluded from the checks that are performed.
2. SinkWriter is an Experimental interface (as of Flink 1.13), all of which are excluded from the checks that are performed.

Test Plan

The Flink codebase has an integration test (FileReadingWatermarkITCase) which involves counting watermarks that arrive at a test sink.   We will update the test code to use the new functionality.

...

1. Combine a process function and a sink function to achieve similar effects with increased complexity.
2. Introduce event-time timers to sink functions.  This approach seems too element-centric for the use cases.   Also, event-time timers are generally keyed.

Future Enhancements

1. Add support for idleness.  A sink operator may become idled due to upstream status changes.  A future enhancement would be to inform the sink function in this situation.