Tweaks prompted by review comments

java-connectors/kafka-connect-common/src/main/java/io/lenses/streamreactor/common/config/base/ConfigMap.java

@@ -27,7 +27,7 @@
 @AllArgsConstructor
 public class ConfigMap {
 
-  private Map<String, Object> wrapped;
+  private final Map<String, Object> wrapped;
 
   /**
    * Retrieves a String property value associated with the given key.

java-connectors/kafka-connect-common/src/main/java/io/lenses/streamreactor/common/config/base/RetryConfig.java

@@ -19,12 +19,30 @@
 import lombok.Builder;
 import lombok.Data;
 
+/**
+ * Configuration class for defining retry behavior.
+ * This class encapsulates settings related to retrying operations, such as the maximum
+ * number of retry attempts and the interval (in milliseconds) between retries.
+ *
+ * Different connector implementations may interpret these settings differently based
+ * on their specific requirements and behavior.
+ */
 @Data
 @Builder
 @AllArgsConstructor
 public class RetryConfig {
 
-  private int numberOfRetries;
-  private long errorRetryInterval;
+  /**
+   * Maximum number of retry attempts allowed.
+   * This value specifies the maximum number of times an operation will be retried
+   * before giving up. A value of 0 indicates no retries will be attempted.
+   */
+  private int retryLimit;
 
+  /**
+   * Interval (in milliseconds) between retry attempts.
+   * This value specifies the time delay between consecutive retry attempts,
+   * measured in milliseconds.
+   */
+  private long retryIntervalMillis;
 }

java-connectors/kafka-connect-common/src/main/java/io/lenses/streamreactor/common/config/base/intf/ConnectionConfig.java

@@ -15,5 +15,10 @@
  */
 package io.lenses.streamreactor.common.config.base.intf;
 
+/**
+ * A marker interface for defining connection configurations across different implementations.
+ * Implementations of this interface signify various types of connection configurations,
+ * providing a unified categorization without requiring shared behavior or properties.
+ */
 public interface ConnectionConfig {
 }

java-connectors/kafka-connect-gcp-common/src/main/java/io/lenses/streamreactor/connect/gcp/common/auth/mode/AuthModeEnum.java → java-connectors/kafka-connect-common/src/main/java/io/lenses/streamreactor/common/config/base/model/ConnectorPrefix.java

@@ -13,23 +13,17 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.lenses.streamreactor.connect.gcp.common.auth.mode;
+package io.lenses.streamreactor.common.config.base.model;
 
-import lombok.NonNull;
+import lombok.AllArgsConstructor;
 
-import java.util.Optional;
+@AllArgsConstructor
+public class ConnectorPrefix {
 
-public enum AuthModeEnum {
-    CREDENTIALS,
-    DEFAULT,
-    FILE,
-    NONE;
+    private final String prefix;
 
-    public static Optional<AuthModeEnum> valueOfCaseInsensitiveOptional(@NonNull String name) {
-        try {
-            return Optional.of(AuthModeEnum.valueOf(name.toUpperCase()));
-        } catch (IllegalArgumentException e) {
-            return Optional.empty();
-        }
+    public String prefixKey(String suffix) {
+        return String.format("%s.%s", prefix, suffix);
     }
+
 }

java-connectors/kafka-connect-common/src/test/java/io/lenses/streamreactor/common/config/base/ConfigMapTest.java

@@ -33,18 +33,15 @@ class ConfigMapTest {
 
     @BeforeEach
     void setUp() {
-        // Create a sample map with test properties
         Map<String, Object> testMap = new HashMap<>();
         testMap.put("username", "user123");
         testMap.put("password", new Password("secret"));
 
-        // Initialize ConfigMap with the test map
         configMap = new ConfigMap(testMap);
     }
 
     @Test
     void testGetString_existingKey_shouldReturnValue() {
-        // Test existing key
         Optional<String> value = configMap.getString("username");
 
         assertTrue(value.isPresent());
@@ -53,15 +50,13 @@ void testGetString_existingKey_shouldReturnValue() {
 
     @Test
     void testGetString_nonExistingKey_shouldReturnEmpty() {
-        // Test non-existing key
         Optional<String> value = configMap.getString("invalidKey");
 
         assertFalse(value.isPresent());
     }
 
     @Test
     void testGetPassword_existingKey_shouldReturnPassword() {
-        // Test existing key for Password type
         Optional<Password> password = configMap.getPassword("password");
 
         assertTrue(password.isPresent());
@@ -70,7 +65,6 @@ void testGetPassword_existingKey_shouldReturnPassword() {
 
     @Test
     void testGetPassword_nonExistingKey_shouldReturnEmpty() {
-        // Test non-existing key for Password type
         Optional<Password> password = configMap.getPassword("invalidKey");
 
         assertFalse(password.isPresent());

java-connectors/kafka-connect-gcp-common/src/main/java/io/lenses/streamreactor/connect/gcp/common/auth/GCPConnectionConfig.java

@@ -22,19 +22,32 @@
 import lombok.Builder;
 import lombok.Data;
 
-import java.util.Optional;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 
 @Data
 @Builder
 @AllArgsConstructor
 public class GCPConnectionConfig  implements ConnectionConfig {
 
-    private Optional<String> projectId;
-    private Optional<String> quotaProjectId;
+    // TODO: These values are duplicated with GCPConfigSettings.  This will be fixed in the next PR.
+    private static final int HTTP_NUM_OF_RETRIES_DEFAULT = 5;
+    private static final long HTTP_ERROR_RETRY_INTERVAL_DEFAULT = 50L;
+
+    @Nullable
+    private String projectId;
+    @Nullable
+    private String quotaProjectId;
+    @Nonnull
     private AuthMode authMode;
-    private Optional<String> host;
-    private RetryConfig httpRetryConfig;
-    private HttpTimeoutConfig timeouts;
+    @Nullable
+    private String host;
+    @Nonnull
+    @Builder.Default
+    private RetryConfig httpRetryConfig = RetryConfig.builder().retryLimit(HTTP_NUM_OF_RETRIES_DEFAULT).retryIntervalMillis(HTTP_ERROR_RETRY_INTERVAL_DEFAULT).build();
+    @Nonnull
+    @Builder.Default
+    private HttpTimeoutConfig timeouts = HttpTimeoutConfig.builder().build();
 
 }
 

java-connectors/kafka-connect-gcp-common/src/main/java/io/lenses/streamreactor/connect/gcp/common/auth/GCPServiceBuilderConfigurer.java

@@ -51,11 +51,11 @@ public class GCPServiceBuilderConfigurer {
           >
   B configure(GCPConnectionConfig config, B builder) throws IOException {
 
-    config.getHost().ifPresent(builder::setHost);
+    Optional.ofNullable(config.getHost()).ifPresent(builder::setHost);
 
-    config.getProjectId().ifPresent(builder :: setProjectId);
+    Optional.ofNullable(config.getProjectId()).ifPresent(builder :: setProjectId);
 
-    config.getQuotaProjectId().ifPresent(builder :: setQuotaProjectId);
+    Optional.ofNullable(config.getQuotaProjectId()).ifPresent(builder :: setQuotaProjectId);
 
     val authMode = config.getAuthMode();
 
@@ -69,12 +69,14 @@ B configure(GCPConnectionConfig config, B builder) throws IOException {
   }
 
   private static Optional<TransportOptions> createTransportOptions(HttpTimeoutConfig timeoutConfig) {
-    if (timeoutConfig.getConnectionTimeout().isPresent() || timeoutConfig.getSocketTimeout().isPresent()) {
+    val connectionTimeout = Optional.ofNullable(timeoutConfig.getConnectionTimeoutMillis());
+    val socketTimeout = Optional.ofNullable(timeoutConfig.getSocketTimeoutMillis());
+    if (connectionTimeout.isPresent() || socketTimeout.isPresent()) {
       HttpTransportOptions.Builder httpTransportOptionsBuilder = HttpTransportOptions.newBuilder();
-      timeoutConfig.getSocketTimeout().ifPresent(sock ->
+      socketTimeout.ifPresent(sock ->
               httpTransportOptionsBuilder.setReadTimeout(sock.intValue())
       );
-      timeoutConfig.getConnectionTimeout().ifPresent(conn ->
+      connectionTimeout.ifPresent(conn ->
               httpTransportOptionsBuilder.setConnectTimeout(conn.intValue())
       );
       return Optional.of(httpTransportOptionsBuilder.build());
@@ -85,9 +87,9 @@ private static Optional<TransportOptions> createTransportOptions(HttpTimeoutConf
   private static RetrySettings createRetrySettings(RetryConfig httpRetryConfig) {
 
     return RetrySettings.newBuilder()
-            .setInitialRetryDelay(Duration.ofMillis(httpRetryConfig.getErrorRetryInterval()))
-            .setMaxRetryDelay(Duration.ofMillis(httpRetryConfig.getErrorRetryInterval() * 5))
-            .setMaxAttempts(httpRetryConfig.getNumberOfRetries())
+            .setInitialRetryDelay(Duration.ofMillis(httpRetryConfig.getRetryIntervalMillis()))
+            .setMaxRetryDelay(Duration.ofMillis(httpRetryConfig.getRetryIntervalMillis() * 5))
+            .setMaxAttempts(httpRetryConfig.getRetryLimit())
             .build();
   }
 }

java-connectors/kafka-connect-gcp-common/src/main/java/io/lenses/streamreactor/connect/gcp/common/auth/HttpTimeoutConfig.java

@@ -19,13 +19,15 @@
 import lombok.Builder;
 import lombok.Data;
 
-import javax.swing.text.html.Option;
+import javax.annotation.Nullable;
 import java.util.Optional;
 
 @Data
 @Builder
 @AllArgsConstructor
 public class HttpTimeoutConfig {
-    private Optional<Long> socketTimeout;
-    private Optional<Long> connectionTimeout;
+    @Nullable
+    private Long socketTimeoutMillis;
+    @Nullable
+    private Long connectionTimeoutMillis;
 }

java-connectors/kafka-connect-gcp-common/src/main/java/io/lenses/streamreactor/connect/gcp/common/auth/mode/AuthMode.java

@@ -18,10 +18,19 @@
 import com.google.auth.Credentials;
 
 import java.io.IOException;
-import java.util.Optional;
 
+/**
+ * Interface representing different authentication modes for GCP connectors.
+ * Implementations of this interface provide methods to obtain Google Cloud Platform (GCP) {@link Credentials}.
+ */
 public interface AuthMode {
-    // Marker interface
+
+    /**
+     * Retrieves the GCP credentials required for authentication.
+     *
+     * @return The GCP {@link Credentials}.
+     * @throws IOException If an I/O error occurs while obtaining credentials.
+     */
     Credentials getCredentials() throws IOException;
 }
 

java-connectors/kafka-connect-gcp-common/src/main/java/io/lenses/streamreactor/connect/gcp/common/auth/mode/DefaultAuthMode.java

@@ -17,9 +17,9 @@
 
 import com.google.auth.Credentials;
 import com.google.auth.oauth2.GoogleCredentials;
+import lombok.EqualsAndHashCode;
 
 import java.io.IOException;
-import java.util.Optional;
 
 /**
  * Default authentication mode without explicit credentials.
@@ -28,11 +28,8 @@
  * The credentials are made available to Cloud Client Libraries and Google API Client Libraries, allowing the code to run seamlessly
  * in both development and production environments without altering the authentication process for Google Cloud services and APIs.
  */
+@EqualsAndHashCode
 public class DefaultAuthMode implements AuthMode {
-    private DefaultAuthMode() {
-    }
-
-    public static AuthMode INSTANCE = new DefaultAuthMode();
 
     @Override
     public Credentials getCredentials() throws IOException {

java-connectors/kafka-connect-gcp-common/src/main/java/io/lenses/streamreactor/connect/gcp/common/auth/mode/FileAuthMode.java

@@ -18,11 +18,9 @@
 import com.google.auth.Credentials;
 import com.google.auth.oauth2.GoogleCredentials;
 import lombok.AllArgsConstructor;
-import lombok.Getter;
 
 import java.io.FileInputStream;
 import java.io.IOException;
-import java.util.Optional;
 
 /**
  * Authentication mode using a json file for credentials.

java-connectors/kafka-connect-gcp-common/src/main/java/io/lenses/streamreactor/connect/gcp/common/auth/mode/NoAuthMode.java

@@ -17,18 +17,13 @@
 
 import com.google.auth.Credentials;
 import com.google.cloud.NoCredentials;
-
-import java.io.IOException;
-import java.util.Optional;
+import lombok.EqualsAndHashCode;
 
 /**
  * Authentication mode indicating no authentication is required.
  */
+@EqualsAndHashCode
 public class NoAuthMode implements AuthMode {
-    private NoAuthMode() {
-    }
-
-    public static AuthMode INSTANCE = new NoAuthMode();
 
     @Override
     public Credentials getCredentials() {