Mailing-List: contact issues-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@flink.apache.org
Date: Thu, 31 Aug 2017 13:36:00 +0000 (UTC)
From: "ASF GitHub Bot (JIRA)" <jira@apache.org>
To: issues@flink.apache.org
Message-ID: <JIRA.13098273.1504014387000.169806.1504186560206@Atlassian.JIRA>
In-Reply-To: <JIRA.13098273.1504014387000@Atlassian.JIRA>
References: <JIRA.13098273.1504014387000@Atlassian.JIRA> <JIRA.13098273.1504014387604@jira-lw-us.apache.org>
Subject: [jira] [Commented] (FLINK-7552) Extend SinkFunction interface with
 SinkContext
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit
archived-at: Thu, 31 Aug 2017 13:36:05 -0000


    [ https://issues.apache.org/jira/browse/FLINK-7552?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16148978#comment-16148978 ] 

ASF GitHub Bot commented on FLINK-7552:
---------------------------------------

Github user rmetzger commented on the issue:

    https://github.com/apache/flink/pull/4616
  
    I see. It is a similar problem we've discussed with the SourceContext. Users only implement it, but don't call it.
    
    I believe in this case, it makes sense to add an exception to Japicmp to allow this change.


> Extend SinkFunction interface with SinkContext
> ----------------------------------------------
>
>                 Key: FLINK-7552
>                 URL: https://issues.apache.org/jira/browse/FLINK-7552
>             Project: Flink
>          Issue Type: Bug
>          Components: DataStream API
>            Reporter: Aljoscha Krettek
>            Assignee: Aljoscha Krettek
>             Fix For: 1.4.0
>
>
> Now that we require Java 8 we can extend the {{SinkFunction}} interface without breaking backwards compatibility. I'm proposing this:
> {code}
> /**
>  * Interface for implementing user defined sink functionality.
>  *
>  * @param <IN> Input type parameter.
>  */
> @Public
> 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(SinkContext, Object)}.
> 	 */
> 	@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.
> 	 *
> 	 * @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();
> 	}
> }
> {code}
> For now, this only allows access to the element timestamp. This would allow us to fix the abomination that is {{FlinkKafkaProducer010}}, which is a hybrid {{SinkFunction}}/{{StreamOperator}} only because it needs access to timestamps.


--
This message was sent by Atlassian JIRA
(v6.4.14#64029)