Clean up classes for PR review

Author: fozzie15

sdks/java/io/kafka/build.gradle

@@ -108,10 +108,6 @@ dependencies {
   testRuntimeOnly library.java.slf4j_jdk14
   testRuntimeOnly project(path: ":runners:direct-java", configuration: "shadow")
   kafkaVersions.each {"kafkaVersion$it.key" "org.apache.kafka:kafka-clients:$it.value"}
-  implementation 'com.google.cloud:google-cloud-secretmanager:2.72.0'
-  testImplementation project(path: ":runners:google-cloud-dataflow-java")
-  testImplementation project(path: ":sdks:java:extensions:google-cloud-platform-core")
-  testImplementation project(path: ":sdks:java:io:google-cloud-platform")
 }
 
 kafkaVersions.each { kv ->

sdks/java/io/kafka/file-aware-factories/src/main/java/org/apache/beam/sdk/io/kafka/file/aware/factories/FileAwareFactoryFn.java

@@ -17,28 +17,9 @@
  */
 package org.apache.beam.sdk.io.kafka.file.aware.factories;
 
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
 import com.google.cloud.secretmanager.v1.AccessSecretVersionResponse;
 import com.google.cloud.secretmanager.v1.SecretManagerServiceClient;
 import com.google.cloud.secretmanager.v1.SecretVersionName;
-import com.google.common.base.Preconditions;
 import java.io.File;
 import java.io.IOException;
 import java.nio.channels.FileChannel;
@@ -57,6 +38,7 @@
 import java.util.regex.Pattern;
 import org.apache.beam.sdk.io.FileSystems;
 import org.apache.beam.sdk.transforms.SerializableFunction;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -230,8 +212,7 @@ protected byte[] getSecret(String secretVersion) {
     }
   }
 
-  protected String processSecret(String originalValue, String secretId, byte[] secretValue)
-      throws RuntimeException {
+  protected String processSecret(String originalValue, String secretId, byte[] secretValue) {
     // By Default, this will return the secret value directly. This function can be overridden by
     // derived classes.
     return new String(secretValue, StandardCharsets.UTF_8);

sdks/java/io/kafka/file-aware-factories/src/main/java/org/apache/beam/sdk/io/kafka/file/aware/factories/KerberosConsumerFactoryFn.java

@@ -59,10 +59,8 @@ public KerberosConsumerFactoryFn(String krb5ConfigGcsPath) {
   @Override
   protected Consumer<byte[], byte[]> createObject(Map<String, Object> config) {
     // This will be called after the config map processing has occurred. Therefore, we know that the
-    // property will have
-    // had it's value replaced with a local directory. We don't need to worry about the GCS prefix
-    // in this case.
-    LOG.info("config when creating the objects: {}", config);
+    // property will have had it's value replaced with a local directory.
+    // We don't need to worry about the GCS prefix in this case.
     try {
       String jaasConfig = (String) config.get(JAAS_CONFIG_PROPERTY);
       String localKeytabPath = "";
@@ -73,8 +71,7 @@ protected Consumer<byte[], byte[]> createObject(Map<String, Object> config) {
       }
 
       // Set the permissions on the file to be as strict as possible for security reasons. The
-      // keytab contains
-      // sensitive information and should be as locked down as possible.
+      // keytab contains sensitive information and should be as locked down as possible.
       Path path = Paths.get(localKeytabPath);
       Set<PosixFilePermission> perms = new HashSet<>();
       perms.add(PosixFilePermission.OWNER_READ);
@@ -100,8 +97,6 @@ protected void downloadAndProcessExtraFiles() throws IOException {
 
           System.setProperty("java.security.krb5.conf", localKrb5ConfPath);
           Configuration.getConfiguration().refresh();
-          LOG.info(
-              "Successfully set and refreshed java.security.krb5.conf to {}", localKrb5ConfPath);
         }
       }
     }
@@ -119,15 +114,13 @@ protected String processSecret(String originalValue, String secretId, byte[] sec
             "Error matching values. Secret was discovered but its value is null");
       }
       currentSecretId = currentSecretId.substring(KEYTAB_SECRET_PREFIX.length());
-      LOG.info("currentSecretId: {} and secretId: {}", currentSecretId, secretId);
       if (!currentSecretId.equals(secretId)) {
         // A sasl.jaas.config can contain multiple keytabs in one string. Therefore, we must assume
         // that there can
         // also be multiple keytab secrets in the same string. If the currently matched secret does
         // not equal
         // the secret that we are processing (passed in via secretId) then we do not want to create
-        // a keytab file and
-        // overwrite it.
+        // a keytab file and overwrite it.
         continue;
       }
       String filename = "kafka-client-" + UUID.randomUUID().toString() + ".keytab";
@@ -141,7 +134,7 @@ protected String processSecret(String originalValue, String secretId, byte[] sec
         }
         Files.write(localFilePath, secretValue);
         if (!new File(localFileString).canRead()) {
-          LOG.info("The file is not readable");
+          LOG.warn("The file is not readable");
         }
         LOG.info("Successfully wrote file to path: {}", localFilePath);
       } catch (IOException e) {
@@ -150,7 +143,6 @@ protected String processSecret(String originalValue, String secretId, byte[] sec
     }
     // if no localFile was created, then we can assume that the secret is meant to be kept as a
     // value.
-    LOG.info("LocalFilestring: {}", localFileString);
     return localFileString.isEmpty()
         ? new String(secretValue, StandardCharsets.UTF_8)
         : localFileString;

sdks/java/io/kafka/file-aware-factories/src/main/java/org/apache/beam/sdk/io/kafka/file/aware/factories/package-info.java

@@ -0,0 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/** ConsumerFactoryFns for file paths that exist in GCS or Google SecretManager. */
+package org.apache.beam.sdk.io.kafka.file.aware.factories;

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

@@ -220,10 +220,6 @@ public void testApplyHappyPathResolvesSecretValue() {
     Map<String, Object> config = new HashMap<>();
     config.put("db.password", originalValue);
 
-    // FIX: Create an anonymous inner class that extends our TestFactoryFn
-    // and overrides the resolveSecret method to return a hardcoded value.
-    // This completely avoids the call to the real getSecret method and its
-    // final class dependencies.
     TestFactoryFn factoryWithMockedSecret =
         new TestFactoryFn() {
           @Override
@@ -252,8 +248,6 @@ public void testApplyFailurePathThrowsExceptionForInvalidSecretFormat() {
     config.put("db.password", "password=" + invalidSecret);
 
     // Act & Assert
-    // The spy will call the real method here, which will throw an exception
-    // because the secret path is not parsable.
     RuntimeException ex = Assert.assertThrows(RuntimeException.class, () -> factory.apply(config));
     Assert.assertEquals(IllegalArgumentException.class, ex.getCause().getClass());
   }

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

@@ -53,7 +53,6 @@ public class KerberosConsumerFactoryFnTest {
 
   @Before
   public void setup() {
-    // Use reflection to reset the static variable before each test for isolation
     try {
       java.lang.reflect.Field field =
           KerberosConsumerFactoryFn.class.getDeclaredField("localKrb5ConfPath");
@@ -64,8 +63,6 @@ public void setup() {
     }
 
     factory = spy(new KerberosConsumerFactoryFn(KRB5_GCS_PATH));
-
-    // Store original system property to restore it after tests
     originalKrb5Conf = System.getProperty("java.security.krb5.conf");
   }
 
@@ -102,8 +99,6 @@ public void testHappyPath() {
             + keytabGcsPath
             + "\" principal=\"user@REALM\";");
 
-    // Use try-with-resources for all necessary static mocks
-
     try (MockedStatic<FileAwareFactoryFn> mockedStaticFactory =
             Mockito.mockStatic(FileAwareFactoryFn.class);
         MockedStatic<Configuration> mockedConfiguration = Mockito.mockStatic(Configuration.class);
@@ -120,7 +115,6 @@ public void testHappyPath() {
           .when(() -> FileAwareFactoryFn.downloadGcsFile(keytabGcsPath, expectedKeytabLocalPath))
           .thenReturn(expectedKeytabLocalPath);
 
-      // Mock other dependencies to prevent side effects
       Configuration mockConf = Mockito.mock(Configuration.class);
       mockedConfiguration.when(Configuration::getConfiguration).thenReturn(mockConf);
       mockedFiles