Filter empty distribution metrics

sdks/python/apache_beam/runners/worker/bundle_processor.py

@@ -1257,7 +1257,18 @@ def monitoring_infos(self):
       all_monitoring_infos_dict.update(
           op.monitoring_infos(transform_id, dict(tag_to_pcollection_id)))
 
-    return list(all_monitoring_infos_dict.values())
+    # Filter distributions that have values of zero counts.
+    filtered_dict = {}
+    for key, value in all_monitoring_infos_dict.items():
+      if value.type != "beam:metrics:distribution_int64:v1":
+        filtered_dict[key] = all_monitoring_infos_dict[key]
+      else:
+        coder = coders.VarIntCoder()
+        count, _, _, _ =   monitoring_infos._decode_distribution(
+          coder, value.payload)
+        if (count > 0):
+          filtered_dict[key] = all_monitoring_infos_dict[key]
+    return list(filtered_dict.values())
 
   def shutdown(self):
     # type: () -> None

sdks/python/apache_beam/runners/worker/bundle_processor_test.py

@@ -314,6 +314,14 @@ def test_can_sample(self):
                   ])
           })
       self.assertEqual(samples, expected)
+
+      # Since data was successfully processed, there should be a
+      # beam:metric:sampled_byte_size:v1 metric
+      self.assertTrue([
+          val for val in processor.monitoring_infos()
+          if val.urn == "beam:metric:sampled_byte_size:v1"
+      ])
+
     finally:
       data_sampler.stop()
 
@@ -382,6 +390,13 @@ def test_can_sample_exceptions(self):
       with self.assertRaisesRegex(RuntimeError, 'expected exception'):
         processor.process_bundle('instruction_id')
 
+      # Since no data was successfully processed, ensure no metrics for
+      # beam:metric:sampled_byte_size:v1 were processed
+      self.assertFalse([
+          val for val in processor.monitoring_infos()
+          if val.urn == "beam:metric:sampled_byte_size:v1"
+      ])
+
       # NOTE: The expected sample comes from the input PCollection. This is very
       # important because there can be coder issues if the sample is put in the
       # wrong PCollection.