[Task]: Fill Runner Metrics support gap in Sources

[Abacn]

What needs to happen?

It is found some runners does not support reporting metrics in

1. BoundedSource.split()

• Direct runner
• non-portable Flink runner
• non-portable Spark runner
• unknown - Samza runner, AttemptedMetrics tests are excluded altogether

Dataflow runner (legacy / runner v2) are supported.

2. BoundedReader.advance()

• Spark Structured Streaming Runner

Notably, portable runners support metrics report in split, as it executes Source as a splittable DoFn, at the point the metrics container is present

This task is created to track these gaps.

Issue Priority

Priority: 2 (default / most normal work should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[Abacn]

By saying unsupported, the following log is seen:

Direct runner:

[Test worker] ERROR org.apache.beam.sdk.metrics.MetricsEnvironment - Unable to update metrics on the current thread. Most likely caused by using metrics outside the managed work-execution thread:
  java.lang.Thread.getStackTrace(Thread.java:1564)
org.apache.beam.sdk.metrics.MetricsEnvironment.getCurrentContainer(MetricsEnvironment.java:140)
org.apache.beam.sdk.metrics.DelegatingCounter.inc(DelegatingCounter.java:76)
org.apache.beam.sdk.metrics.DelegatingCounter.inc(DelegatingCounter.java:67)
org.apache.beam.sdk.metrics.MetricsTest$CountingSourceWithMetrics.split(MetricsTest.java:495)
org.apache.beam.runners.direct.BoundedReadEvaluatorFactory$InputProvider.getInitialInputs(BoundedReadEvaluatorFactory.java:217)
org.apache.beam.runners.direct.ReadEvaluatorFactory$InputProvider.getInitialInputs(ReadEvaluatorFactory.java:88)
org.apache.beam.runners.direct.RootProviderRegistry.getInitialInputs(RootProviderRegistry.java:80)
org.apache.beam.runners.direct.ExecutorServiceParallelExecutor.start(ExecutorServiceParallelExecutor.java:161)
org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:213)
...

Flink runner:

[jobmanager-io-thread-1] ERROR org.apache.beam.sdk.metrics.MetricsEnvironment - Unable to update metrics on the current thread. Most likely caused by using metrics outside the managed work-execution thread:
  java.lang.Thread.getStackTrace(Thread.java:1564)
org.apache.beam.sdk.metrics.MetricsEnvironment.getCurrentContainer(MetricsEnvironment.java:140)
org.apache.beam.sdk.metrics.DelegatingCounter.inc(DelegatingCounter.java:76)
org.apache.beam.sdk.metrics.DelegatingCounter.inc(DelegatingCounter.java:67)
org.apache.beam.sdk.metrics.MetricsTest$CountingSourceWithMetrics.split(MetricsTest.java:495)
org.apache.beam.runners.flink.translation.wrappers.SourceInputFormat.createInputSplits(SourceInputFormat.java:135)
org.apache.beam.runners.flink.translation.wrappers.SourceInputFormat.createInputSplits(SourceInputFormat.java:44)
org.apache.flink.runtime.executiongraph.ExecutionJobVertex.initialize(ExecutionJobVertex.java:251)
org.apache.flink.runtime.executiongraph.DefaultExecutionGraph.initializeJobVertex(DefaultExecutionGraph.java:894)
org.apache.flink.runtime.executiongraph.ExecutionGraph.initializeJobVertex(ExecutionGraph.java:224)
...

This is different from a log ususally seen when reporting metrics in a callback thread

beam/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricsEnvironment.java

Line 144 in 89d5e2f

"Reporting metrics are not supported in the current execution environment:\n {}",

here Java core's isMetricsSupported() returns true, but not actually supported by the runner. So this is a feature gap rather than WAI by spec