thanks. it worked!
I add the following method at the
org.apache.flink.streaming.api.operators.StreamingRuntimeContext
class:
public Environment getTaskEnvironment() { return this.taskEnvironment; }
Then I am getting the IP using:
ConfigOption<String> restAddressOption = ConfigOptions
.key("rest.address")
.stringType()
.noDefaultValue();
String restAddress =
this.getRuntimeContext().getTaskEnvironment().getTaskManagerInfo().getConfiguration().getValue(restAddressOption);
Thanks!
--
-- Felipe Gutierrez
-- skype: felipe.o.gutierrez
-- https://felipeogutierrez.blogspot.com
On Fri, May 22, 2020 at 3:54 AM Yangze Guo <karma...@gmail.com> wrote:
>
> 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
