Hi,
Thank you for your reply. What I want is flink app has multiple jobs, each
job manage a stream. Currently our flink app has only 1 job that manage
multiple streams.
I did try env.executeAsyc(), but it still doesn't work. From the log, when
the second executeAsync() was called, it shows " *Job
00000000000000000000000000000000 was recovered successfully.*"
Looks like the second executeAsync() recover the first job. Not start a
second job.
Thanks,
Qihua
On Thu, Jun 17, 2021 at 7:43 AM Arvid Heise <ar...@apache.org> wrote:
> Hi,
>
> env.execute("Job 1"); is a blocking call. You either have to use
> executeAsync or use a separate thread to submit the second job. If Job 1
> finishes then this would also work by having sequential execution.
>
> However, I think what you actually want to do is to use the same env with
> 2 topologies and 1 single execute like this.
>
> StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> DataStream<String> stream1 = env.addSource(new SourceFunction<String>());
> stream1.addSink(new FlinkKafkaProducer());
> DataStream<String> stream2 = env.addSource(new SourceFunction<String>());
> stream2.addSink(new DiscardingSink<>());
> env.execute("Job 1+2");
>
> On Wed, Jun 16, 2021 at 6:04 PM Qihua Yang <yang...@gmail.com> wrote:
>
>> Hi,
>> Does anyone know how to run multiple jobs in same flink application?
>> I did a simple test. First job was started. I did see the log message,
>> but I didn't see the second job was started, even I saw the log message.
>>
>> public void testJobs() throws Exception {
>> StreamExecutionEnvironment env =
>> StreamExecutionEnvironment.getExecutionEnvironment();
>> DataStream<String> stream1 = env.addSource(new SourceFunction<String>());
>> stream1.addSink(new FlinkKafkaProducer());
>> printf("### first job");
>> env.execute("Job 1");
>>
>> env = StreamExecutionEnvironment.getExecutionEnvironment();
>> DataStream<String> stream2 = env.addSource(new SourceFunction<String>());
>> stream2.addSink(new DiscardingSink<>());
>> printf("### second job");
>> env.execute("Job 2");
>> }
>>
>> Here is the log:
>> ### first job
>> INFO o.a.f.c.deployment.application.executors.EmbeddedExecutor - Job
>> 00000000000000000000000000000000 is submitted.
>> INFO o.a.f.c.deployment.application.executors.EmbeddedExecutor -
>> Submitting Job with JobId=00000000000000000000000000000000.
>> INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher -
>> Received JobGraph submission 00000000000000000000000000000000 (job1).
>> INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher -
>> Submitting job 00000000000000000000000000000000 (job1).
>>
>> INFO o.a.f.r.leaderretrieval.ZooKeeperLeaderRetrievalService - Starting
>> ZooKeeperLeaderRetrievalService /leader/resource_manager_lock.
>> INFO org.apache.flink.runtime.jobmaster.JobMaster - Starting execution
>> of job job1 (00000000000000000000000000000000) under job master id
>> b03cde9dc2aebdb39c46cda4c2a94c07.
>> INFO org.apache.flink.runtime.jobmaster.JobMaster - Starting scheduling
>> with scheduling strategy
>> [org.apache.flink.runtime.scheduler.strategy.EagerSchedulingStrategy]
>> INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Job job1
>> (00000000000000000000000000000000) switched from state CREATED to RUNNING.
>>
>> ### second job
>> WARN com.doordash.flink.common.utils.LoggerUtil - Class - IndexWriter :
>> ### second job
>> INFO org.apache.flink.runtime.jobmaster.JobMaster - Resolved
>> ResourceManager address, beginning registration
>> INFO o.a.f.r.leaderretrieval.ZooKeeperLeaderRetrievalService - Starting
>> ZooKeeperLeaderRetrievalService
>> /leader/00000000000000000000000000000000/job_manager_lock.
>> INFO o.a.flink.runtime.resourcemanager.StandaloneResourceManager -
>> Registering job manager b03cde9dc2aebdb39c46cda4c2a94...@akka.tcp://
>> flink@10.4.139.110:6123/user/rpc/jobmanager_2 for job
>> 00000000000000000000000000000000.
>> INFO o.a.f.c.deployment.application.executors.EmbeddedExecutor - Job
>> 00000000000000000000000000000000 was recovered successfully.
>> INFO o.a.flink.runtime.resourcemanager.StandaloneResourceManager -
>> Registered job manager b03cde9dc2aebdb39c46cda4c2a94...@akka.tcp://
>> flink@10.4.139.110:6123/user/rpc/jobmanager_2 for job
>> 00000000000000000000000000000000.
>> INFO org.apache.flink.runtime.jobmaster.JobMaster - JobManager
>> successfully registered at ResourceManager, leader id:
>> 956d4431ca90d45d92c027046cd0404e.
>> INFO org.apache.flink.runtime.jobmaster.slotpool.SlotPoolImpl -
>> Requesting new slot [SlotRequestId{8980dce0c0ef6a933d73051c58534489}] and
>> profile ResourceProfile{UNKNOWN} from resource manager.
>> INFO o.a.flink.runtime.resourcemanager.StandaloneResourceManager -
>> Request slot with profile ResourceProfile{UNKNOWN} for job
>> 00000000000000000000000000000000 with allocation id
>> 21134414fc60d4ef3e940609cef960b6.
>> INFO org.apache.flink.runtime.jobmaster.slotpool.SlotPoolImpl -
>> Requesting new slot [SlotRequestId{e37cc684758e49f3cce76c4687b1d1a3}] and
>> profile ResourceProfile{UNKNOWN} from resource manager.
>> INFO o.a.flink.runtime.resourcemanager.StandaloneResourceManager -
>> Request slot with profile ResourceProfile{UNKNOWN} for job
>> 00000000000000000000000000000000 with allocation id
>> 650bd9100a35ef5086fd4614f5253b55.
>>
>
