Hi, Felipe
I see your problem. IIUC, if you use AbstractUdfStreamOperator, you
could indeed get all the configurations(including what you defined in
flink-conf.yaml) through
"AbstractUdfStreamOperator#getRuntimeContext().getTaskManagerRuntimeInfo().getConfiguration()".
However, I guess it is not the right behavior and might be fixed in
future versions.
Best,
Yangze Guo
On Thu, May 21, 2020 at 3:13 PM Felipe Gutierrez
<felipe.o.gutier...@gmail.com> wrote:
>
> Hi all,
>
> I would like to have the IP of the JobManager, not the Task Executors.
> I explain why.
>
> I have an operator (my own operator that extends
> AbstractUdfStreamOperator) that sends and receives messages from a
> global controller. So, regardless of which TaskManager these operator
> instances are deployed, they need to send and receive messages from my
> controller. Currently, I am doing this using MQTT broker (this is my
> first approach and I don't know if there is a better way to do it,
> maybe there is...)
>
> The first thing that I do is to start my controller using the
> org.apache.flink.runtime.jobmaster.JobManagerRunnerImpl and subscribe
> it to the JobManager host. I am getting the IP of the JobManager by
> adding this method on the
> org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory
> class:
> public String getRpcServiceAddress() {
> return this.rpcService.getAddress();
> }
> That is working. Although I am not sure if it is the best approach.
>
> The second thing that I am doing is to make each operator instance
> publish and subscribe to this controller. To do this they need the
> JobManager IP. I could get the TaskManager IPs from the
> AbstractUdfStreamOperator, but not the JobManager IP. So, I am passing
> the JobManager IP as a parameter to the operator at the moment. I
> suppose that it is easy to get the JobManager IP inside the
> AbstractUdfStreamOperator or simply add some method somewhere to get
> this value. However, I don't know where.
>
> Thanks,
> Felipe
>
> --
> -- Felipe Gutierrez
> -- skype: felipe.o.gutierrez
> -- https://felipeogutierrez.blogspot.com
>
> On Thu, May 21, 2020 at 7:13 AM Yangze Guo <karma...@gmail.com> wrote:
> >
> > Hi, Felipe
> >
> > Do you mean to get the Host and Port of the task executor where your
> > operator is indeed running on?
> >
> > If that is the case, IIUC, two possible components that contain this
> > information are RuntimeContext and the Configuration param of
> > RichFunction#open. After reading the relevant code path, it seems you
> > could not get it at the moment.
> >
> > Best,
> > Yangze Guo
> >
> > Best,
> > Yangze Guo
> >
> >
> > On Wed, May 20, 2020 at 11:46 PM Alexander Fedulov
> > <alexan...@ververica.com> wrote:
> > >
> > > Hi Felippe,
> > >
> > > could you clarify in some more details what you are trying to achieve?
> > >
> > > Best regards,
> > >
> > > --
> > >
> > > Alexander Fedulov | Solutions Architect
> > >
> > > +49 1514 6265796
> > >
> > >
> > >
> > > Follow us @VervericaData
> > >
> > > --
> > >
> > > Join Flink Forward - The Apache Flink Conference
> > >
> > > Stream Processing | Event Driven | Real Time
> > >
> > > --
> > >
> > > Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
> > >
> > > --
> > >
> > > Ververica GmbH
> > > Registered at Amtsgericht Charlottenburg: HRB 158244 B
> > > Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
> > > (Tony) Cheng
> > >
> > >
> > >
> > >
> > > On Wed, May 20, 2020 at 1:14 PM Felipe Gutierrez
> > > <felipe.o.gutier...@gmail.com> wrote:
> > >>
> > >> Hi all,
> > >>
> > >> I have my own operator that extends the AbstractUdfStreamOperator
> > >> class and I want to issue some messages to it. Sometimes the operator
> > >> instances are deployed on different TaskManagers and I would like to
> > >> set some attributes like the master and slave IPs on it.
> > >>
> > >> I am trying to use these values but they only return localhost, not
> > >> the IP configured at flink-conf.yaml file. (jobmanager.rpc.address:
> > >> 192.168.56.1).
> > >>
> > >> ConfigOption<String> restAddressOption = ConfigOptions
> > >> .key("rest.address")
> > >> .stringType()
> > >> .noDefaultValue();
> > >> System.out.println("DefaultJobManagerRunnerFactory rest.address: " +
> > >> jobMasterConfiguration.getConfiguration().getValue(restAddressOption));
> > >> System.out.println("rpcService: " + rpcService.getAddress());
> > >>
> > >>
> > >> Thanks,
> > >> Felipe
> > >>
> > >> --
> > >> -- Felipe Gutierrez
> > >> -- skype: felipe.o.gutierrez
> > >> -- https://felipeogutierrez.blogspot.com
