add option to disable metrics (#34303)

Co-authored-by: Naireen <[email protected]>

Author: Naireen

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIOInitializer.java

@@ -19,13 +19,16 @@
 
 import com.google.auto.service.AutoService;
 import org.apache.beam.sdk.harness.JvmInitializer;
+import org.apache.beam.sdk.options.ExperimentalOptions;
 import org.apache.beam.sdk.options.PipelineOptions;
 
 /** Initialize KafkaIO feature flags on worker. */
 @AutoService(JvmInitializer.class)
 public class KafkaIOInitializer implements JvmInitializer {
   @Override
   public void beforeProcessing(PipelineOptions options) {
-    KafkaSinkMetrics.setSupportKafkaMetrics(true);
+    if (!ExperimentalOptions.hasExperiment(options, "disable_kafka_metrics")) {
+      KafkaSinkMetrics.setSupportKafkaMetrics(true);
+    }
   }
 }