junaiddshaukat commented on code in PR #39341:
URL: https://github.com/apache/beam/pull/39341#discussion_r3588504231
##########
runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/KafkaStreamsTestRunner.java:
##########
@@ -117,6 +123,8 @@ public static void run(Pipeline pipeline) {
driver.advanceWallClockTime(Duration.ofSeconds(1));
roundTripInternalTopics(driver, internalTopics(topology));
}
+ return MetricsContainerStepMap.asAttemptedOnlyMetricResults(
+ context.getMetricsContainerStepMap());
Review Comment:
Same as the first thread — the shared container is safe; see reply there.
##########
runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPortablePipelineResult.java:
##########
@@ -104,8 +110,9 @@ public State waitUntilFinish() {
@Override
public MetricResults metrics() {
- throw new UnsupportedOperationException(
- "Metrics are not yet implemented in the Kafka Streams runner.");
+ // Attempted values only: the runner does not distinguish committed
results yet (that needs
+ // metrics to be folded into the exactly-once commit, which lands with the
durability work).
+ return
MetricsContainerStepMap.asAttemptedOnlyMetricResults(metricsContainerStepMap);
}
Review Comment:
Same as the first thread — the shared container is safe; see reply there.
##########
runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPortablePipelineResult.java:
##########
@@ -41,11 +42,16 @@ class KafkaStreamsPortablePipelineResult implements
PortablePipelineResult {
LoggerFactory.getLogger(KafkaStreamsPortablePipelineResult.class);
private final KafkaStreams kafkaStreams;
+ // The job's metrics accumulator, shared by reference with the topology's
stage processors, which
+ // update it as the SDK harness reports bundle metrics.
+ private final MetricsContainerStepMap metricsContainerStepMap;
private final CountDownLatch terminated = new CountDownLatch(1);
private volatile boolean cancelled = false;
- KafkaStreamsPortablePipelineResult(KafkaStreams kafkaStreams) {
+ KafkaStreamsPortablePipelineResult(
+ KafkaStreams kafkaStreams, MetricsContainerStepMap
metricsContainerStepMap) {
this.kafkaStreams = kafkaStreams;
+ this.metricsContainerStepMap = metricsContainerStepMap;
Review Comment:
Same as the first thread — the shared container is safe; see reply there.
##########
runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPortablePipelineResult.java:
##########
@@ -21,6 +21,7 @@
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.beam.model.jobmanagement.v1.JobApi;
+import org.apache.beam.runners.core.metrics.MetricsContainerStepMap;
import org.apache.beam.runners.jobsubmission.PortablePipelineResult;
import org.apache.beam.sdk.metrics.MetricResults;
Review Comment:
Same as the first thread — the shared container is safe; see reply there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
