RichAsyncFunction in scala

[Antoine Philippot]

Hi,

It lacks a version of RichAsyncFunction class in the scala API or the possibility to handle a class which extends AbstractRichFunction and implements AsyncFunction (from the scala API).

I made a small dev on our current flink fork because we need to use the open method to add our custom metrics from getRuntimeContext.getMetricGroup method.

https://github.com/aphilippot/flink/commit/acbd49cf2d64163040f2f954ce55917155b408ba

Do you already plan to release this feature soon ? Do you want me to create a new Jira ticket, propose a pull request ?

Antoine

[Ufuk Celebi]

Hey Antoine,

isn't it possible to use the Java RichAsyncFunction from Scala like this:

class Test extends RichAsyncFunction[Int, Int] {

  override def open(parameters: Configuration): Unit = super.open(parameters)

  override def asyncInvoke(input: Int, resultFuture:
functions.async.ResultFuture[Int]): Unit = ???
}

– Ufuk



On Thu, Dec 28, 2017 at 4:37 PM, Antoine Philippot
<[hidden email]> wrote:

> Hi,
>
> It lacks a version of RichAsyncFunction class in the scala API or the
> possibility to handle a class which extends AbstractRichFunction and
> implements AsyncFunction (from the scala API).
>
> I made a small dev on our current flink fork because we need to use the open
> method to add our custom metrics from getRuntimeContext.getMetricGroup
> method.
> https://github.com/aphilippot/flink/commit/acbd49cf2d64163040f2f954ce55917155b408ba
>
> Do you already plan to release this feature soon ? Do you want me to create
> a new Jira ticket, propose a pull request ?
>
> Antoine

[Antoine Philippot]

Hi Ufuk,

I don't think it is possible as I use this function as a parameter of AsyncDataStream (from the scala API) which is mandatory to use with the scala DataStream.

Le jeu. 28 déc. 2017 à 16:55, Ufuk Celebi <[hidden email]> a écrit :

Hey Antoine,

isn't it possible to use the Java RichAsyncFunction from Scala like this:

class Test extends RichAsyncFunction[Int, Int] {

  override def open(parameters: Configuration): Unit = super.open(parameters)

  override def asyncInvoke(input: Int, resultFuture:
functions.async.ResultFuture[Int]): Unit = ???
}

– Ufuk



On Thu, Dec 28, 2017 at 4:37 PM, Antoine Philippot
<[hidden email]> wrote:
> Hi,
>
> It lacks a version of RichAsyncFunction class in the scala API or the
> possibility to handle a class which extends AbstractRichFunction and
> implements AsyncFunction (from the scala API).
>
> I made a small dev on our current flink fork because we need to use the open
> method to add our custom metrics from getRuntimeContext.getMetricGroup
> method.
> https://github.com/aphilippot/flink/commit/acbd49cf2d64163040f2f954ce55917155b408ba
>
> Do you already plan to release this feature soon ? Do you want me to create
> a new Jira ticket, propose a pull request ?
>
> Antoine

[Aljoscha Krettek]

Hi,

There is this Jira issue: https://issues.apache.org/jira/browse/FLINK-6756

Best,

Aljoscha

On 28. Dec 2017, at 17:10, Antoine Philippot <[hidden email]> wrote:

Hi Ufuk,

I don't think it is possible as I use this function as a parameter of AsyncDataStream (from the scala API) which is mandatory to use with the scala DataStream.

Le jeu. 28 déc. 2017 à 16:55, Ufuk Celebi <[hidden email]> a écrit :

Hey Antoine,

isn't it possible to use the Java RichAsyncFunction from Scala like this:

class Test extends RichAsyncFunction[Int, Int] {

  override def open(parameters: Configuration): Unit = super.open(parameters)

  override def asyncInvoke(input: Int, resultFuture:
functions.async.ResultFuture[Int]): Unit = ???
}

– Ufuk



On Thu, Dec 28, 2017 at 4:37 PM, Antoine Philippot
<[hidden email]> wrote:
> Hi,
>
> It lacks a version of RichAsyncFunction class in the scala API or the
> possibility to handle a class which extends AbstractRichFunction and
> implements AsyncFunction (from the scala API).
>
> I made a small dev on our current flink fork because we need to use the open
> method to add our custom metrics from getRuntimeContext.getMetricGroup
> method.
> https://github.com/aphilippot/flink/commit/acbd49cf2d64163040f2f954ce55917155b408ba
>
> Do you already plan to release this feature soon ? Do you want me to create
> a new Jira ticket, propose a pull request ?
>
> Antoine