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

@@ -27,7 +27,6 @@
 import io.confluent.kafka.serializers.KafkaAvroDeserializer;
 import java.io.InputStream;
 import java.io.OutputStream;
-import java.lang.reflect.Constructor;
 import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -36,6 +35,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;
@@ -96,7 +96,6 @@
 import org.apache.beam.sdk.transforms.splittabledofn.WatermarkEstimators.MonotonicallyIncreasing;
 import org.apache.beam.sdk.transforms.splittabledofn.WatermarkEstimators.WallTime;
 import org.apache.beam.sdk.util.InstanceBuilder;
-import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
 import org.apache.beam.sdk.util.Preconditions;
 import org.apache.beam.sdk.util.construction.PTransformMatchers;
 import org.apache.beam.sdk.util.construction.ReplacementOutputs;
@@ -937,18 +936,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 +1102,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

@@ -110,24 +110,19 @@
 
 ReadFromKafkaSchema = typing.NamedTuple(
     'ReadFromKafkaSchema',
-    [
-        ('consumer_config', typing.Mapping[str, str]),
-        ('topics', typing.List[str]),
-        ('key_deserializer', str),
-        ('value_deserializer', str),
-        ('start_read_time', typing.Optional[int]),
-        ('max_num_records', typing.Optional[int]),
-        ('max_read_time', typing.Optional[int]),
-        ('commit_offset_in_finalize', bool),
-        ('timestamp_policy', str),
-        ('consumer_polling_timeout', typing.Optional[int]),
-        ('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]),
-        ('consumer_factory_fn_params', typing.Optional[typing.Mapping[str, str]])
-    ])
+    [('consumer_config', typing.Mapping[str, str]),
+     ('topics', typing.List[str]), ('key_deserializer', str),
+     ('value_deserializer', str), ('start_read_time', typing.Optional[int]),
+     ('max_num_records', typing.Optional[int]),
+     ('max_read_time', typing.Optional[int]),
+     ('commit_offset_in_finalize', bool), ('timestamp_policy', str),
+     ('consumer_polling_timeout', typing.Optional[int]),
+     ('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_class', typing.Optional[str]),
+     ('consumer_factory_fn_params', typing.Optional[typing.Mapping[str, str]])])
 
 
 def default_io_expansion_service(append_args=None):
@@ -176,8 +171,7 @@ def __init__(
       allow_duplicates=False,
       dynamic_read_poll_interval_seconds: typing.Optional[int] = None,
       consumer_factory_fn_class=None,
-      consumer_factory_fn_params=None
-  ):
+      consumer_factory_fn_params=None):
     """
     Initializes a read operation from Kafka.
 
@@ -220,13 +214,13 @@ def __init__(
     :param dynamic_read_poll_interval_seconds: The interval in seconds at which
         to check for new partitions. If not None, dynamic partition discovery
         is enabled.
-    :param consumer_factory_fn_class: A fully qualified classpath to an existing provided
-        consumerFactoryFn. If not None, this will construct Kafka consumers with a custom
-        configuration.
-    :param consumer_factory_fn_params: A map which specifies the parameters for the provided
-        consumer_factory_fn_class. IF not None, the values in this map will be used when
-        constructing the consumer_factory_fn_class object. This cannot be null
-        if the consumer_factory_fn_class is not null.
+    :param consumer_factory_fn_class: A fully qualified classpath to an
+        existing provided consumerFactoryFn. If not None, this will construct
+        Kafka consumers with a custom configuration.
+    :param consumer_factory_fn_params: A map which specifies the parameters for
+        the provided consumer_factory_fn_class. IF not None, the values in this
+        map will be used when constructing the consumer_factory_fn_class object.
+        This cannot be null if the consumer_factory_fn_class is not null.
     """
     if timestamp_policy not in [ReadFromKafka.processing_time_policy,
                                 ReadFromKafka.create_time_policy,