Migrate to a new module such that kafka remains GCP Agnostic.

Author: fozzie15

build.gradle.kts

@@ -349,6 +349,7 @@ tasks.register("javaioPreCommit") {
   dependsOn(":sdks:java:io:jms:build")
   dependsOn(":sdks:java:io:kafka:build")
   dependsOn(":sdks:java:io:kafka:upgrade:build")
+  dependsOn(":sdks:java:io:kafka:file-aware-factories:build")
   dependsOn(":sdks:java:io:kudu:build")
   dependsOn(":sdks:java:io:mongodb:build")
   dependsOn(":sdks:java:io:mqtt:build")

sdks/java/io/kafka/file-aware-factories/build.gradle

@@ -0,0 +1,42 @@
+/*
+ * 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.
+ */
+
+plugins { id 'org.apache.beam.module' }
+applyJavaNature(
+        automaticModuleName: 'org.apache.beam.sdk.io.kafka.file.aware.factories',
+        publish: 'False'
+)
+
+description = "Apache Beam :: SDKs :: Java :: IO :: Kafka :: File :: Aware :: Factories"
+ext.summary = "Library to instantiate kafka clients with files from GCS or SecretManager."
+
+dependencies {
+    // ------------------------- CORE DEPENDENCIES -------------------------
+    implementation project(path: ":sdks:java:core", configuration: "shadow")
+    provided library.java.kafka_clients
+    implementation 'com.google.cloud:google-cloud-secretmanager:2.72.0'
+    implementation library.java.slf4j_api
+    implementation library.java.vendored_guava_32_1_2_jre
+
+    // ------------------------- TEST DEPENDENCIES -------------------------
+    testImplementation 'org.apache.kafka:kafka-clients:3.9.0'
+    testImplementation library.java.junit
+    testImplementation library.java.mockito_core
+    testRuntimeOnly library.java.mockito_inline
+    testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
+}

…eam/sdk/io/kafka/FileAwareFactoryFn.java …/aware/factories/FileAwareFactoryFn.javasdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/FileAwareFactoryFn.java renamed to sdks/java/io/kafka/file-aware-factories/src/main/java/org/apache/beam/sdk/io/kafka/file/aware/factories/FileAwareFactoryFn.java sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/FileAwareFactoryFn.java renamed to sdks/java/io/kafka/file-aware-factories/src/main/java/org/apache/beam/sdk/io/kafka/file/aware/factories/FileAwareFactoryFn.java

@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.sdk.io.kafka;
+package org.apache.beam.sdk.io.kafka.file.aware.factories;
 
 /*
  * Licensed to the Apache Software Foundation (ASF) under one
@@ -196,9 +196,9 @@ protected byte[] getSecretWithCache(String secretId) {
    * @return a string with all instances of GCS paths converted to the local paths where the files
    *     sit.
    */
-    private String replacePathWithLocal(String gcsPath) throws IOException {
-      return DIRECTORY_PREFIX + "/" + factoryType + "/" + gcsPath.substring(GCS_PATH_PREFIX.length());
-    }
+  private String replacePathWithLocal(String gcsPath) throws IOException {
+    return DIRECTORY_PREFIX + "/" + factoryType + "/" + gcsPath.substring(GCS_PATH_PREFIX.length());
+  }
 
   /**
    * @throws IOException A hook for subclasses to download and process specific files before the

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

@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.sdk.io.kafka;
+package org.apache.beam.sdk.io.kafka.file.aware.factories;
 
 import java.io.File;
 import java.io.IOException;

…sdk/io/kafka/FileAwareFactoryFnTest.java …re/factories/FileAwareFactoryFnTest.javasdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/FileAwareFactoryFnTest.java renamed to sdks/java/io/kafka/file-aware-factories/src/test/java/org/apache/beam/sdk/io/kafka/file/aware/factories/FileAwareFactoryFnTest.java sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/FileAwareFactoryFnTest.java renamed to sdks/java/io/kafka/file-aware-factories/src/test/java/org/apache/beam/sdk/io/kafka/file/aware/factories/FileAwareFactoryFnTest.java

@@ -15,18 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.sdk.io.kafka;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThrows;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.mockStatic;
-import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.when;
+package org.apache.beam.sdk.io.kafka.file.aware.factories;
 
 import java.io.ByteArrayInputStream;
 import java.io.File;
@@ -42,13 +31,16 @@
 import org.apache.beam.sdk.io.FileSystems;
 import org.apache.beam.sdk.io.fs.MatchResult;
 import org.apache.beam.sdk.io.fs.ResourceId;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
+import org.mockito.ArgumentMatchers;
 import org.mockito.MockedStatic;
+import org.mockito.Mockito;
 
 @RunWith(JUnit4.class)
 public class FileAwareFactoryFnTest {
@@ -75,8 +67,8 @@ protected Object createObject(Map<String, Object> config) {
   @Before
   public void setup() throws IOException {
     baseDir = "/tmp/" + TEST_FACTORY_TYPE;
-    factory = spy(new TestFactoryFn());
-    doReturn(baseDir).when(factory).getBaseDirectory();
+    factory = Mockito.spy(new TestFactoryFn());
+    Mockito.doReturn(baseDir).when(factory).getBaseDirectory();
   }
 
   @Test
@@ -93,11 +85,11 @@ public void testHappyPathReplacesGcsPath() {
 
     // Act & Assert
     // Use try-with-resources to manage the scope of the static mock on FileSystems
-    try (MockedStatic<FileSystems> mockedFileSystems = mockStatic(FileSystems.class)) {
+    try (MockedStatic<FileSystems> mockedFileSystems = Mockito.mockStatic(FileSystems.class)) {
       // 1. Mock the underlying static FileSystems calls to avoid real network I/O
-      MatchResult.Metadata metadata = mock(MatchResult.Metadata.class);
-      ResourceId resourceId = mock(ResourceId.class);
-      when(metadata.resourceId()).thenReturn(resourceId);
+      MatchResult.Metadata metadata = Mockito.mock(MatchResult.Metadata.class);
+      ResourceId resourceId = Mockito.mock(ResourceId.class);
+      Mockito.when(metadata.resourceId()).thenReturn(resourceId);
       mockedFileSystems.when(() -> FileSystems.matchSingleFileSpec(gcsPath)).thenReturn(metadata);
 
       // 2. Mock 'open' to return a channel with no data, simulating a successful download
@@ -108,8 +100,9 @@ public void testHappyPathReplacesGcsPath() {
       Map<String, Object> processedConfig = (Map<String, Object>) factory.apply(config);
 
       // Assert
-      assertEquals(expectedLocalPath, processedConfig.get("config.file.path"));
-      assertTrue("Local file should have been created", new File(expectedLocalPath).exists());
+      Assert.assertEquals(expectedLocalPath, processedConfig.get("config.file.path"));
+      Assert.assertTrue(
+          "Local file should have been created", new File(expectedLocalPath).exists());
     }
   }
 
@@ -121,17 +114,17 @@ public void testApplyFailurePathThrowsRuntimeExceptionOnDownloadFailure() {
     config.put("critical.file", gcsPath);
 
     // Mock the static FileSystems.matchSingleFileSpec to throw an exception
-    try (MockedStatic<FileSystems> mockedFileSystems = mockStatic(FileSystems.class)) {
+    try (MockedStatic<FileSystems> mockedFileSystems = Mockito.mockStatic(FileSystems.class)) {
       mockedFileSystems
           .when(() -> FileSystems.matchSingleFileSpec(gcsPath))
           .thenThrow(new IOException("GCS file not found"));
 
       // Act & Assert
       RuntimeException exception =
-          assertThrows(RuntimeException.class, () -> factory.apply(config));
-      assertTrue(exception.getMessage().contains("Failed trying to process value"));
-      assertTrue(exception.getCause() instanceof IOException);
-      assertTrue(exception.getCause().getMessage().contains("Failed to download file"));
+          Assert.assertThrows(RuntimeException.class, () -> factory.apply(config));
+      Assert.assertTrue(exception.getMessage().contains("Failed trying to process value"));
+      Assert.assertTrue(exception.getCause() instanceof IOException);
+      Assert.assertTrue(exception.getCause().getMessage().contains("Failed to download file"));
     }
   }
 
@@ -147,7 +140,7 @@ public void testApplyHappyPathIgnoresNonGcsValues() {
     Map<String, Object> processedConfig = (Map<String, Object>) factory.apply(config);
 
     // Assert
-    assertEquals(config, processedConfig);
+    Assert.assertEquals(config, processedConfig);
   }
 
   @Test
@@ -172,7 +165,7 @@ public void testApplyEdgeCaseMultipleGcsPathsInSingleValue() {
     Map<String, Object> config = new HashMap<>();
     config.put("jaas.config", originalValue);
 
-    try (MockedStatic<FileSystems> mockedFileSystems = mockStatic(FileSystems.class)) {
+    try (MockedStatic<FileSystems> mockedFileSystems = Mockito.mockStatic(FileSystems.class)) {
       // Mock GCS calls for both paths
       mockSuccessfulDownload(mockedFileSystems, gcsPath1);
       mockSuccessfulDownload(mockedFileSystems, gcsPath2);
@@ -181,7 +174,7 @@ public void testApplyEdgeCaseMultipleGcsPathsInSingleValue() {
       Map<String, Object> processedConfig = (Map<String, Object>) factory.apply(config);
 
       // Assert
-      assertEquals(expectedProcessedValue, processedConfig.get("jaas.config"));
+      Assert.assertEquals(expectedProcessedValue, processedConfig.get("jaas.config"));
     }
   }
 
@@ -193,22 +186,25 @@ public void testApplyEdgeCaseLocalFileWriteFails() throws IOException {
     config.put("a.file", gcsPath);
 
     // Mock GCS part to succeed
-    try (MockedStatic<FileSystems> mockedFileSystems = mockStatic(FileSystems.class);
-        MockedStatic<FileChannel> mockedFileChannel = mockStatic(FileChannel.class)) {
+    try (MockedStatic<FileSystems> mockedFileSystems = Mockito.mockStatic(FileSystems.class);
+        MockedStatic<FileChannel> mockedFileChannel = Mockito.mockStatic(FileChannel.class)) {
       mockSuccessfulDownload(mockedFileSystems, gcsPath);
 
       // Mock the local file writing part to fail
       mockedFileChannel
-          .when(() -> FileChannel.open(any(Path.class), any(Set.class)))
+          .when(
+              () ->
+                  FileChannel.open(
+                      ArgumentMatchers.any(Path.class), ArgumentMatchers.any(Set.class)))
           .thenThrow(new IOException("Permission denied"));
 
       // Act & Assert
       RuntimeException exception =
-          assertThrows(RuntimeException.class, () -> factory.apply(config));
-      assertTrue(exception.getMessage().contains("Failed trying to process value"));
-      assertTrue(exception.getCause() instanceof IOException);
+          Assert.assertThrows(RuntimeException.class, () -> factory.apply(config));
+      Assert.assertTrue(exception.getMessage().contains("Failed trying to process value"));
+      Assert.assertTrue(exception.getCause() instanceof IOException);
       // Check that the root cause is our "Permission denied" mock
-      assertTrue(exception.getCause().getCause().getMessage().contains("Permission denied"));
+      Assert.assertTrue(exception.getCause().getCause().getMessage().contains("Permission denied"));
     }
   }
 
@@ -233,7 +229,7 @@ public void testApplyHappyPathResolvesSecretValue() {
           @Override
           public byte[] getSecret(String secretIdentifier) {
             // Assert that the correct identifier is passed
-            assertEquals(secretVersionParsed, secretIdentifier);
+            Assert.assertEquals(secretVersionParsed, secretIdentifier);
             // Return a predictable, hardcoded value for the test
             return secretValue.getBytes(StandardCharsets.UTF_8);
           }
@@ -245,7 +241,7 @@ public byte[] getSecret(String secretIdentifier) {
         (Map<String, Object>) factoryWithMockedSecret.apply(config);
 
     // Assert
-    assertEquals(expectedProcessedValue, processedConfig.get("db.password"));
+    Assert.assertEquals(expectedProcessedValue, processedConfig.get("db.password"));
   }
 
   @Test
@@ -258,18 +254,22 @@ public void testApplyFailurePathThrowsExceptionForInvalidSecretFormat() {
     // Act & Assert
     // The spy will call the real method here, which will throw an exception
     // because the secret path is not parsable.
-    RuntimeException ex = assertThrows(RuntimeException.class, () -> factory.apply(config));
-    assertEquals(IllegalArgumentException.class, ex.getCause().getClass());
+    RuntimeException ex = Assert.assertThrows(RuntimeException.class, () -> factory.apply(config));
+    Assert.assertEquals(IllegalArgumentException.class, ex.getCause().getClass());
   }
 
   // Helper method to reduce boilerplate in mocking successful GCS downloads
   private void mockSuccessfulDownload(MockedStatic<FileSystems> mockedFileSystems, String gcsPath) {
-    MatchResult.Metadata metadata = mock(MatchResult.Metadata.class);
-    ResourceId resourceId = mock(ResourceId.class);
-    when(metadata.resourceId()).thenReturn(resourceId);
-    mockedFileSystems.when(() -> FileSystems.matchSingleFileSpec(eq(gcsPath))).thenReturn(metadata);
+    MatchResult.Metadata metadata = Mockito.mock(MatchResult.Metadata.class);
+    ResourceId resourceId = Mockito.mock(ResourceId.class);
+    Mockito.when(metadata.resourceId()).thenReturn(resourceId);
+    mockedFileSystems
+        .when(() -> FileSystems.matchSingleFileSpec(ArgumentMatchers.eq(gcsPath)))
+        .thenReturn(metadata);
 
     ReadableByteChannel channel = Channels.newChannel(new ByteArrayInputStream(new byte[0]));
-    mockedFileSystems.when(() -> FileSystems.open(eq(resourceId))).thenReturn(channel);
+    mockedFileSystems
+        .when(() -> FileSystems.open(ArgumentMatchers.eq(resourceId)))
+        .thenReturn(channel);
   }
 }

…kafka/KerberosConsumerFactoryFnTest.java …ories/KerberosConsumerFactoryFnTest.javasdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KerberosConsumerFactoryFnTest.java renamed to sdks/java/io/kafka/file-aware-factories/src/test/java/org/apache/beam/sdk/io/kafka/file/aware/factories/KerberosConsumerFactoryFnTest.java sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KerberosConsumerFactoryFnTest.java renamed to sdks/java/io/kafka/file-aware-factories/src/test/java/org/apache/beam/sdk/io/kafka/file/aware/factories/KerberosConsumerFactoryFnTest.java

@@ -15,17 +15,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.beam.sdk.io.kafka;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.mockConstruction;
-import static org.mockito.Mockito.mockStatic;
+package org.apache.beam.sdk.io.kafka.file.aware.factories;
+
 import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.verify;
 
 import java.io.File;
 import java.io.IOException;
@@ -40,13 +32,16 @@
 import javax.security.auth.login.Configuration;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 import org.mockito.ArgumentCaptor;
+import org.mockito.ArgumentMatchers;
 import org.mockito.MockedConstruction;
 import org.mockito.MockedStatic;
+import org.mockito.Mockito;
 
 @RunWith(JUnit4.class)
 public class KerberosConsumerFactoryFnTest {
@@ -110,12 +105,13 @@ public void testHappyPath() {
     // Use try-with-resources for all necessary static mocks
 
     try (MockedStatic<FileAwareFactoryFn> mockedStaticFactory =
-            mockStatic(FileAwareFactoryFn.class);
-        MockedStatic<Configuration> mockedConfiguration = mockStatic(Configuration.class);
-        MockedStatic<Files> mockedFiles = mockStatic(Files.class);
-        MockedConstruction<KafkaConsumer> mockedConsumer = mockConstruction(KafkaConsumer.class)) {
+            Mockito.mockStatic(FileAwareFactoryFn.class);
+        MockedStatic<Configuration> mockedConfiguration = Mockito.mockStatic(Configuration.class);
+        MockedStatic<Files> mockedFiles = Mockito.mockStatic(Files.class);
+        MockedConstruction<KafkaConsumer> mockedConsumer =
+            Mockito.mockConstruction(KafkaConsumer.class)) {
 
-      assertNotNull(mockedConsumer);
+      Assert.assertNotNull(mockedConsumer);
       // Mock the static downloadGcsFile method to prevent any GCS interaction
       mockedStaticFactory
           .when(() -> FileAwareFactoryFn.downloadGcsFile(KRB5_GCS_PATH, expectedKrb5LocalPath))
@@ -125,29 +121,34 @@ public void testHappyPath() {
           .thenReturn(expectedKeytabLocalPath);
 
       // Mock other dependencies to prevent side effects
-      Configuration mockConf = mock(Configuration.class);
+      Configuration mockConf = Mockito.mock(Configuration.class);
       mockedConfiguration.when(Configuration::getConfiguration).thenReturn(mockConf);
       mockedFiles
-          .when(() -> Files.setPosixFilePermissions(any(Path.class), any(Set.class)))
+          .when(
+              () ->
+                  Files.setPosixFilePermissions(
+                      ArgumentMatchers.any(Path.class), ArgumentMatchers.any(Set.class)))
+          .thenReturn(null);
+      mockedFiles
+          .when(() -> Files.createDirectories(ArgumentMatchers.any(Path.class)))
           .thenReturn(null);
-      mockedFiles.when(() -> Files.createDirectories(any(Path.class))).thenReturn(null);
 
       // Act
       factory.apply(config);
 
       // Assert
       // 1. Verify that the krb5.conf system property was set correctly.
-      assertEquals(expectedKrb5LocalPath, System.getProperty("java.security.krb5.conf"));
+      Assert.assertEquals(expectedKrb5LocalPath, System.getProperty("java.security.krb5.conf"));
 
       // 2. Capture the config passed to createObject and verify the keytab path was replaced.
       ArgumentCaptor<Map<String, Object>> configCaptor = ArgumentCaptor.forClass(Map.class);
-      verify(factory).createObject(configCaptor.capture());
+      Mockito.verify(factory).createObject(configCaptor.capture());
       Map<String, Object> capturedConfig = configCaptor.getValue();
       String processedJaasConfig = (String) capturedConfig.get("sasl.jaas.config");
-      assertTrue(processedJaasConfig.contains("keyTab=\"" + expectedKeytabLocalPath + "\""));
+      Assert.assertTrue(processedJaasConfig.contains("keyTab=\"" + expectedKeytabLocalPath + "\""));
 
       // 3. Verify that the JAAS configuration was refreshed.
-      verify(mockConf).refresh();
+      Mockito.verify(mockConf).refresh();
     }
   }
 }