rohitsinha54 commented on code in PR #32068:
URL: https://github.com/apache/beam/pull/32068#discussion_r1705805792
##########
sdks/java/io/google-cloud-platform/build.gradle:
##########
@@ -218,6 +218,10 @@ task integrationTest(type: Test, dependsOn:
processTestResources) {
useJUnit {
excludeCategories "org.apache.beam.sdk.testing.UsesKms"
+ filter {
+ // https://github.com/apache/beam/issues/32071
Review Comment:
interesting.....
Thanks for catching this.
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableReadIT.java:
##########
@@ -138,6 +144,17 @@ public void testE2EBigtableSegmentRead() {
.withMaxBufferElementCount(10))
.apply(Count.globally());
PAssert.thatSingleton(count).isEqualTo(numRows);
- p.run();
+ PipelineResult r = p.run();
+ checkLineageSourceMetric(r, tableId);
+ }
+
+ private void checkLineageSourceMetric(PipelineResult r, String tableId) {
+ // TODO(https://github.com/apache/beam/issues/32071) test malformed,
+ // when pipeline.run() os non-blocking, the metrics are not available by
the time of query
Review Comment:
typo "is"
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Lineage.java:
##########
@@ -38,4 +38,35 @@ public static StringSet getSources() {
public static StringSet getSinks() {
return SINKS;
}
+
+ /** Query {@link StringSet} metrics from {@link MetricResults}. */
+ public static Set<String> query(MetricResults results, Type type) {
Review Comment:
Nice!
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableServiceImpl.java:
##########
@@ -71,6 +71,7 @@
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO.BigtableSource;
import org.apache.beam.sdk.io.range.ByteKeyRange;
import org.apache.beam.sdk.metrics.Distribution;
+import org.apache.beam.sdk.metrics.Lineage;
Review Comment:
Thank you.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java:
##########
@@ -2017,8 +2030,11 @@ public boolean start() throws IOException {
&& rangeTracker.tryReturnRecordAt(
true, makeByteKey(reader.getCurrentRow().getKey())))
|| rangeTracker.markDone();
+ LOG.warn("called start");
Review Comment:
remove log lines
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Lineage.java:
##########
@@ -23,11 +23,8 @@
public class Lineage {
public static final String LINEAGE_NAMESPACE = "lineage";
- public static final String SOURCE_METRIC_NAME = "sources";
Review Comment:
SG. Thanks.
--
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]
