You are viewing a plain text version of this content. The canonical link for it is here.

Posted to user@flink.apache.org by Debasish Ghosh <gh...@gmail.com> on 2019/05/15 07:30:07 UTC

Passing a custom SourceContext to a SourceFunction

Hi -

I have a custom SourceFunction ..

class MySourceFunction[T](data: Seq[T]) extends SourceFunction[T] {
  def run(ctx: SourceContext[T]): Unit = {
    data.foreach(d ⇒ ctx.collect(d))
  }
}

When this function is run during job execution, the SourceContext that gets
passed serializes the data. I would like to pass a mock SourceContext
(similar to
https://github.com/apache/flink/blob/master/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/ListSourceContext.java)
in the run method. How do I do this ? Note I am not invoking the run method
explicitly anywhere.

Any help will be appreciated.

regards.

-- 
Debasish Ghosh
http://manning.com/ghosh2
http://manning.com/ghosh

Twttr: @debasishg
Blog: http://debasishg.blogspot.com
Code: http://github.com/debasishg

Re: Passing a custom SourceContext to a SourceFunction

Posted by Chesnay Schepler <ch...@apache.org>.

You cannot control what kind of SourceContext is passed into your function.

What are you trying to achieve?

On 15/05/2019 09:30, Debasish Ghosh wrote:
> Hi -
>
> I have a custom SourceFunction ..
>
> class MySourceFunction[T](data: Seq[T]) extends SourceFunction[T] {
>   def run(ctx: SourceContext[T]): Unit = {
>     data.foreach(d ⇒ ctx.collect(d))
>   }
> }
>
> When this function is run during job execution, the SourceContext that 
> gets passed serializes the data. I would like to pass a mock 
> SourceContext (similar to 
> https://github.com/apache/flink/blob/master/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/ListSourceContext.java) 
> in the run method. How do I do this ? Note I am not invoking the run 
> method explicitly anywhere.
>
> Any help will be appreciated.
>
> regards.
>
> -- 
> Debasish Ghosh
> http://manning.com/ghosh2
> http://manning.com/ghosh
>
> Twttr: @debasishg
> Blog: http://debasishg.blogspot.com
> Code: http://github.com/debasishg