[Bug]: Flink runner fails with "The job graph is cyclic."

### What happened?

I tried a extremely simple beam pipeline:
```
package main

import (
"context"
"flag"
"github.com/apache/beam/sdks/v2/go/pkg/beam"
"github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
"log"
)

func main() {
flag.Parse()
beam.Init()

ctx := context.Background()
p, s := beam.NewPipelineWithRoot()

beam.Impulse(s)

if err := beamx.Run(ctx, p); err != nil {
log.Fatal(ctx, "Failed to execute job: %v", err)
}
}
```

When using the flink runner with the following error:

```
C:\Users\waj33\AppData\Local\Temp\GoLand\___2Telemetry_Beam_Pipeline.exe --runner=flink --endpoint=localhost:8099

2022/06/18 19:29:36 No environment config specified. Using default config: 'apache/beam_go_sdk:2.41.0.dev'
2022/06/18 19:29:36 components: <
transforms: <
key: "e1"
value: <
unique_name: "Impulse"
spec: <
urn: "beam:transform:impulse:v1"
>
outputs: <
key: "i0"
value: "n1"
>
>
>
pcollections: <
key: "n1"
value: <
unique_name: "n1"
coder_id: "c0"
is_bounded: BOUNDED
windowing_strategy_id: "w0"
>
>
windowing_strategies: <
key: "w0"
value: <
window_fn: <
urn: "beam:window_fn:global_windows:v1"
>
merge_status: NON_MERGING
window_coder_id: "c1"
trigger: <
default: <
>
>
accumulation_mode: DISCARDING
output_time: END_OF_WINDOW
closing_behavior: EMIT_IF_NONEMPTY
on_time_behavior: FIRE_IF_NONEMPTY
environment_id: "go"
>
>
coders: <
key: "c0"
value: <
spec: <
urn: "beam:coder:bytes:v1"
>
>
>
coders: <
key: "c1"
value: <
spec: <
urn: "beam:coder:global_window:v1"
>
>
>
environments: <
key: "go"
value: <
urn: "beam:env:docker:v1"
payload: "\n\035apache/beam_go_sdk:2.41.0.dev"
capabilities: "beam:protocol:progress_reporting:v1"
capabilities: "beam:protocol:multi_core_bundle_processing:v1"
capabilities: "beam:transform:sdf_truncate_sized_restrictions:v1"
capabilities: "beam:protocol:worker_status:v1"
capabilities: "beam:protocol:monitoring_info_short_ids:v1"
capabilities: "beam:version:sdk_base:go"
capabilities: "beam:coder:bytes:v1"
capabilities: "beam:coder:bool:v1"
capabilities: "beam:coder:varint:v1"
capabilities: "beam:coder:double:v1"
capabilities: "beam:coder:string_utf8:v1"
capabilities: "beam:coder:length_prefix:v1"
capabilities: "beam:coder:kv:v1"
capabilities: "beam:coder:iterable:v1"
capabilities: "beam:coder:state_backed_iterable:v1"
capabilities: "beam:coder:windowed_value:v1"
capabilities: "beam:coder:global_window:v1"
capabilities: "beam:coder:interval_window:v1"
capabilities: "beam:coder:row:v1"
capabilities: "beam:coder:nullable:v1"
dependencies: <
type_urn: "beam:artifact:type:file:v1"
role_urn: "beam:artifact:role:go_worker_binary:v1"
>
>
>
>
root_transform_ids: "e1"
2022/06/18 19:29:39 Prepared job with id: go-job-1-1655598576054300700_3356a1e1-23e9-4522-8839-f5547d5b18cb and staging token: go-job-1-1655598576054300700_3356a1e1-23e9-4522-8839-f5547d5b18cb
2022/06/18 19:29:39 Staged binary artifact with token:
2022/06/18 19:29:39 Submitted job: go0job0101655598576054300700-root-0619002939-c5a32241_10511e7a-30ce-4f6e-872a-5023c28ba4d9
2022/06/18 19:29:39 Job state: STOPPED
2022/06/18 19:29:39 Job state: STARTING
2022/06/18 19:29:39 Job state: RUNNING
2022/06/18 19:29:45 (): java.lang.RuntimeException: java.util.concurrent.ExecutionException: java.lang.RuntimeException: org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster.
at org.apache.flink.util.ExceptionUtils.rethrow(ExceptionUtils.java:316)
at org.apache.flink.api.java.ExecutionEnvironment.executeAsync(ExecutionEnvironment.java:1061)
at org.apache.flink.api.java.ExecutionEnvironment.execute(ExecutionEnvironment.java:958)
at org.apache.beam.runners.flink.FlinkBatchPortablePipelineTranslator$BatchTranslationContext.execute(FlinkBatchPortablePipelineTranslator.java:195)
at org.apache.beam.runners.flink.FlinkPipelineRunner.runPipelineWithTranslator(FlinkPipelineRunner.java:118)
at org.apache.beam.runners.flink.FlinkPipelineRunner.run(FlinkPipelineRunner.java:85)
at org.apache.beam.runners.jobsubmission.JobInvocation.runPipeline(JobInvocation.java:86)
at org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:125)
at org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:57)
at org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:78)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
Caused by: java.util.concurrent.ExecutionException: java.lang.RuntimeException: org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster.
at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908)
at org.apache.flink.api.java.ExecutionEnvironment.executeAsync(ExecutionEnvironment.java:1056)
... 11 more
Caused by: java.lang.RuntimeException: org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster.
at org.apache.flink.util.ExceptionUtils.rethrow(ExceptionUtils.java:316)
at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedFunction$2(FunctionUtils.java:75)
at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616)
at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591)
at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:456)
... 1 more
Caused by: org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster.
at org.apache.flink.runtime.jobmaster.DefaultJobMasterServiceProcess.lambda$new$0(DefaultJobMasterServiceProcess.java:97)
at java.util.concurrent.CompletableFuture.uniWhenComplete(Unknown Source)
at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(Unknown Source)
at java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.lang.Thread.run(Unknown Source)
Caused by: java.util.concurrent.CompletionException: org.apache.flink.api.common.InvalidProgramException: The job graph is cyclic.
at java.util.concurrent.CompletableFuture.encodeThrowable(Unknown Source)
at java.util.concurrent.CompletableFuture.completeThrowable(Unknown Source)
... 4 more
Caused by: org.apache.flink.api.common.InvalidProgramException: The job graph is cyclic.
at org.apache.flink.runtime.jobgraph.JobGraph.getVerticesSortedTopologicallyFromSources(JobGraph.java:442)
at org.apache.flink.runtime.executiongraph.DefaultExecutionGraphBuilder.buildGraph(DefaultExecutionGraphBuilder.java:186)
at org.apache.flink.runtime.scheduler.DefaultExecutionGraphFactory.createAndRestoreExecutionGraph(DefaultExecutionGraphFactory.java:149)
at org.apache.flink.runtime.scheduler.SchedulerBase.createAndRestoreExecutionGraph(SchedulerBase.java:363)
at org.apache.flink.runtime.scheduler.SchedulerBase.(SchedulerBase.java:208)
at org.apache.flink.runtime.scheduler.DefaultScheduler.(DefaultScheduler.java:191)
at org.apache.flink.runtime.scheduler.DefaultScheduler.(DefaultScheduler.java:139)
at org.apache.flink.runtime.scheduler.DefaultSchedulerFactory.createInstance(DefaultSchedulerFactory.java:135)
at org.apache.flink.runtime.jobmaster.DefaultSlotPoolServiceSchedulerFactory.createScheduler(DefaultSlotPoolServiceSchedulerFactory.java:115)
at org.apache.flink.runtime.jobmaster.JobMaster.createScheduler(JobMaster.java:345)
at org.apache.flink.runtime.jobmaster.JobMaster.(JobMaster.java:322)
at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.internalCreateJobMasterService(DefaultJobMasterServiceFactory.java:106)
at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.lambda$createJobMasterService$0(DefaultJobMasterServiceFactory.java:94)
at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedSupplier$4(FunctionUtils.java:112)
... 4 more
2022/06/18 19:29:45 (): org.apache.flink.api.common.InvalidProgramException: The job graph is cyclic.
2022/06/18 19:29:45 Job state: FAILED
2022/06/18 19:29:45 context.BackgroundFailed to execute job: %vjob go0job0101655598576054300700-root-0619002939-c5a32241_10511e7a-30ce-4f6e-872a-5023c28ba4d9 failed

Process finished with the exit code 1
```

### Issue Priority

Priority: 3

### Issue Component

Component: sdk-go

Start with the minimal Go pipeline using beam.Impulse and beamx.Run, then reproduce it with the Flink runner at localhost:8099. Trace how this pipeline is translated into the Flink job graph; done means the pipeline starts successfully without the cyclic-job-graph error.

Written by the indexing model from the issue text.