Remove accidentally committed python modules

Author: fozzie15

sdks/java/extensions/kafka-factories/src/test/java/org/apache/beam/sdk/io/kafka/file/aware/factories/FileAwareFactoryFnTest.java

@@ -36,6 +36,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.regex.Pattern;
@@ -114,6 +115,7 @@
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Joiner;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -937,18 +939,27 @@ static <K, V> void setupExternalBuilder(
         if (config.consumerFactoryFnClass != null) {
           if (config.consumerFactoryFnClass.contains("KerberosConsumerFactoryFn")) {
             try {
-              if (!config.consumerFactoryFnClass.contains("krb5Location")) {
-                throw new IllegalArgumentException("The KerberosConsumerFactoryFn requires a location for the krb5.conf file. " +
-                    "Please provide either a GCS location or Google Secret Manager location for this file.");
+              if (!config.consumerFactoryFnParams.containsKey("krb5Location")) {
+                throw new IllegalArgumentException(
+                    "The KerberosConsumerFactoryFn requires a location for the krb5.conf file. "
+                        + "Please provide either a GCS location or Google Secret Manager location for this file.");
               }
-                builder.setConsumerFactoryFn(InstanceBuilder.ofType(new TypeDescriptor<SerializableFunction<Map<String, Object>, Consumer<byte[], byte[]>>>() {})
-                    .fromClassName(config.consumerFactoryFnClass)
-                    .withArg(
-                            String.class,
-                            config.consumerFactoryFnParams.get("krb5Location"))
-                    .build());
+              String krb5Location = config.consumerFactoryFnParams.get("krb5Location");
+              builder.setConsumerFactoryFn(
+                  InstanceBuilder.ofType(
+                          new TypeDescriptor<
+                              SerializableFunction<
+                                  Map<String, Object>, Consumer<byte[], byte[]>>>() {})
+                      .fromClassName(config.consumerFactoryFnClass)
+                      .withArg(String.class, Objects.requireNonNull(krb5Location))
+                      .build());
             } catch (Exception e) {
-              throw new RuntimeException("Unable to construct FactoryFn " + config.consumerFactoryFnClass + ": " + e.getMessage(), e);
+              throw new RuntimeException(
+                  "Unable to construct FactoryFn "
+                      + config.consumerFactoryFnClass
+                      + ": "
+                      + e.getMessage(),
+                  e);
             }
           }
         }
@@ -1094,12 +1105,12 @@ public void setDynamicReadPollIntervalSeconds(Long dynamicReadPollIntervalSecond
         }
 
         public void setConsumerFactoryFnClass(String consumerFactoryFnClass) {
-            this.consumerFactoryFnClass = consumerFactoryFnClass;
+          this.consumerFactoryFnClass = consumerFactoryFnClass;
         }
 
         public void setConsumerFactoryFnParams(Map<String, String> consumerFactoryFnParams) {
-            this.consumerFactoryFnParams = consumerFactoryFnParams;
-          }
+          this.consumerFactoryFnParams = consumerFactoryFnParams;
+        }
       }
     }
 

sdks/java/extensions/kafka-factories/src/test/java/org/apache/beam/sdk/io/kafka/file/aware/factories/KerberosConsumerFactoryFnTest.java

@@ -41,10 +41,6 @@
 from apache_beam.transforms.userstate import BagStateSpec
 from apache_beam.transforms.userstate import CombiningValueStateSpec
 from apache_beam.utils import subprocess_server
-from apache_beam.options.pipeline_options import (
-    PipelineOptions,
-    GoogleCloudOptions
-)
 
 NUM_RECORDS = 1000
 
@@ -72,7 +68,7 @@ def process(
 
 class CrossLanguageKafkaIO(object):
   def __init__(
-      self, bootstrap_servers=None, topic=None, null_key=None, expansion_service=None):
+      self, bootstrap_servers, topic, null_key, expansion_service=None):
     self.bootstrap_servers = bootstrap_servers
     self.topic = topic
     self.null_key = null_key
@@ -111,34 +107,6 @@ def build_read_pipeline(self, pipeline, max_num_records=None):
         | 'CalculateSum' >> beam.ParDo(CollectingFn())
         | 'SetSumCounter' >> beam.Map(self.sum_counter.inc))
 
-  def build_read_pipeline_with_kerberos(self, p, max_num_records=None):
-
-      jaas_config = (
-          f'com.sun.security.auth.module.Krb5LoginModule required '
-          f'useKeyTab=true storeKey=true '
-          f'keyTab="secretValue:projects/dataflow-testing-311516/secrets/kafka-client-keytab/versions/latest" '
-          f'principal="[email protected]";'
-      )
-
-      kafka_records = (
-          p
-          | 'ReadFromKafka' >> ReadFromKafka(
-              consumer_config={
-                  'bootstrap.servers': 'fozzie-test-kafka-broker.us-central1-c.c.dataflow-testing-311516.internal:9092',
-                  'auto.offset.reset': 'earliest',
-                  'max_num_records': max_num_records,
-                  'security.protocol': 'SASL_PLAINTEXT',
-                  'sasl.mechanism': 'GSSAPI',
-                  'sasl.kerberos.service.name': 'kafka',
-                  'sasl.jaas.config': jaas_config
-              },
-              topics=['fozzie_test_kerberos_topic'],
-              key_deserializer='org.apache.kafka.common.serialization.StringDeserializer',
-              value_deserializer='org.apache.kafka.common.serialization.StringDeserializer',
-              consumer_factory_fn_class='org.apache.beam.sdk.extensions.kafka.factories.KerberosConsumerFactoryFn',
-              consumer_factory_fn_params={'krb5Location': 'gs://fozzie_testing_bucket/kerberos/krb5.conf'}))
-      return kafka_records
-
   def run_xlang_kafkaio(self, pipeline):
     self.build_write_pipeline(pipeline)
     self.build_read_pipeline(pipeline)
@@ -214,17 +182,6 @@ def test_hosted_kafkaio_null_key(self):
     self.run_kafka_write(pipeline_creator)
     self.run_kafka_read(pipeline_creator, None)
 
-  def test_hosted_kafkaio_null_key_kerberos(self):
-      kafka_topic = 'xlang_kafkaio_test_null_key_{}'.format(uuid.uuid4())
-      bootstrap_servers = 'fozzie-test-kafka-broker.us-central1-c.c.dataflow-testing-311516.internal:9092'
-      pipeline_creator = CrossLanguageKafkaIO(
-          bootstrap_servers,
-          kafka_topic,
-          True,
-          'localhost:%s' % os.environ.get('EXPANSION_PORT'))
-
-      self.run_kafka_read_with_kerberos(pipeline_creator)
-
   def run_kafka_write(self, pipeline_creator):
     with TestPipeline() as pipeline:
       pipeline.not_use_test_runner_api = True
@@ -239,23 +196,6 @@ def run_kafka_read(self, pipeline_creator, expected_key):
           equal_to([(expected_key, str(i).encode())
                     for i in range(NUM_RECORDS)]))
 
-  def run_kafka_read_with_kerberos(self, pipeline_creator):
-      options_dict = {
-          'runner': 'DataflowRunner',
-          'project': 'dataflow-testing-311516',
-          'region': 'us-central1',
-          'streaming': False
-      }
-      options = PipelineOptions.from_dictionary(options_dict)
-      expected_records = [f'test{i}' for i in range(1, 12)]
-      with beam.Pipeline(options=options) as p:
-          pipeline.not_use_test_runner_api = True
-          result = pipeline_creator.build_read_pipeline_with_kerberos(p, max_num_records=11)
-          assert_that(
-              result,
-              equal_to(expected_records)
-          )
-
   def get_platform_localhost(self):
     if sys.platform == 'darwin':
       return 'host.docker.internal'

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

@@ -124,8 +124,8 @@
         ('redistribute', typing.Optional[bool]),
         ('redistribute_num_keys', typing.Optional[np.int32]),
         ('allow_duplicates', typing.Optional[bool]),
-        ('dynamic_read_poll_interval_seconds', typing.Optional[int])
-        ('consumer_factory_fn', typing.Optional[str]),
+        ('dynamic_read_poll_interval_seconds', typing.Optional[int]),
+        ('consumer_factory_fn_class', typing.Optional[str]),
         ('consumer_factory_fn_params', typing.Optional[typing.Mapping[str, str]])
     ])