Make gcs.credentials.default ternary

`null` should be the default value for `gcs.credentials.default`, because otherwise certain practical combinations are impossible (e.g. just setting `gcs.credentials.json` without anything else).

Also this commit refactors how the credentials settings are validated: the validation logic is completely delegated to `CredentialsBuilder` to have it in one place.

storage/gcs/src/main/java/io/aiven/kafka/tieredstorage/storage/gcs/CredentialsBuilder.java

@@ -22,6 +22,8 @@
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Paths;
+import java.util.Objects;
+import java.util.stream.Stream;
 
 import com.google.auth.Credentials;
 import com.google.auth.oauth2.GoogleCredentials;
@@ -32,12 +34,26 @@ private CredentialsBuilder() {
         // hide constructor
     }
 
+    public static void validate(final Boolean defaultCredentials,
+                                final String credentialsJson,
+                                final String credentialsPath) {
+        final long nonNulls = Stream.of(defaultCredentials, credentialsJson, credentialsPath)
+            .filter(Objects::nonNull).count();
+        if (nonNulls == 0) {
+            throw new IllegalArgumentException(
+                "All defaultCredentials, credentialsJson, and credentialsPath cannot be null simultaneously.");
+        }
+        if (nonNulls > 1) {
+            throw new IllegalArgumentException(
+                "Only one of defaultCredentials, credentialsJson, and credentialsPath can be non-null.");
+        }
+    }
+
     /**
      * Builds {@link GoogleCredentials}.
      *
-     * <p>{@code credentialsPath} and {@code credentialsJson} are mutually exclusive: if both are provided (are
-     * non-{@code null}), this is an error. They are also cannot be set together with
-     * {@code defaultCredentials == true}.
+     * <p>{@code credentialsPath}, {@code credentialsJson}, and {@code defaultCredentials true}
+     * are mutually exclusive: if more than one are provided (are non-{@code null}), this is an error.
      *
      * <p>If either {@code credentialsPath} or {@code credentialsJson} is provided, it's used to
      * construct the credentials.
@@ -52,26 +68,11 @@ private CredentialsBuilder() {
      * @throws IOException              if some error getting the credentials happen.
      * @throws IllegalArgumentException if a combination of parameters is invalid.
      */
-    public static Credentials build(final boolean defaultCredentials,
+    public static Credentials build(final Boolean defaultCredentials,
                                     final String credentialsJson,
                                     final String credentialsPath)
         throws IOException {
-        if (defaultCredentials && credentialsJson != null) {
-            throw new IllegalArgumentException(
-                "defaultCredentials == true and credentialsJson != null cannot be simultaneously.");
-        }
-        if (defaultCredentials && credentialsPath != null) {
-            throw new IllegalArgumentException(
-                "defaultCredentials == true and credentialsPath != null cannot be simultaneously.");
-        }
-        if (credentialsJson != null && credentialsPath != null) {
-            throw new IllegalArgumentException(
-                "Both credentialsPath and credentialsJson cannot be non-null.");
-        }
-
-        if (defaultCredentials) {
-            return GoogleCredentials.getApplicationDefault();
-        }
+        validate(defaultCredentials, credentialsJson, credentialsPath);
 
         if (credentialsJson != null) {
             return getCredentialsFromJson(credentialsJson);
@@ -81,7 +82,11 @@ public static Credentials build(final boolean defaultCredentials,
             return getCredentialsFromPath(credentialsPath);
         }
 
-        return NoCredentials.getInstance();
+        if (Boolean.TRUE.equals(defaultCredentials)) {
+            return GoogleCredentials.getApplicationDefault();
+        } else {
+            return NoCredentials.getInstance();
+        }
     }
 
     private static GoogleCredentials getCredentialsFromPath(final String credentialsPath) throws IOException {

storage/gcs/src/main/java/io/aiven/kafka/tieredstorage/storage/gcs/GcsStorageConfig.java

@@ -99,7 +99,7 @@ class GcsStorageConfig extends AbstractConfig {
             .define(
                 GCP_CREDENTIALS_DEFAULT_CONFIG,
                 ConfigDef.Type.BOOLEAN,
-                true,
+                null,
                 ConfigDef.Importance.MEDIUM,
                 GCP_CREDENTIALS_DEFAULT_DOC);
     }
@@ -110,28 +110,22 @@ public GcsStorageConfig(final Map<String, ?> props) {
     }
 
     private void validate() {
-        if (getBoolean(GCP_CREDENTIALS_DEFAULT_CONFIG)
-            && getPassword(GCP_CREDENTIALS_JSON_CONFIG) != null) {
-            throw new ConfigException(GCP_CREDENTIALS_JSON_CONFIG
-                + " and "
-                + GCP_CREDENTIALS_DEFAULT_CONFIG
-                + " cannot be set together");
-        }
+        final String credentialsJson = getPassword(GCP_CREDENTIALS_JSON_CONFIG) == null
+            ? null
+            : getPassword(GCP_CREDENTIALS_JSON_CONFIG).value();
 
-        if (getBoolean(GCP_CREDENTIALS_DEFAULT_CONFIG)
-            && getString(GCP_CREDENTIALS_PATH_CONFIG) != null) {
-            throw new ConfigException(GCP_CREDENTIALS_PATH_CONFIG
-                + " and "
-                + GCP_CREDENTIALS_DEFAULT_CONFIG
-                + " cannot be set together");
-        }
-
-        if (getPassword(GCP_CREDENTIALS_JSON_CONFIG) != null
-            && getString(GCP_CREDENTIALS_PATH_CONFIG) != null) {
-            throw new ConfigException(GCP_CREDENTIALS_JSON_CONFIG
-                + " and "
-                + GCP_CREDENTIALS_PATH_CONFIG
-                + " cannot be set together");
+        try {
+            CredentialsBuilder.validate(
+                getBoolean(GCP_CREDENTIALS_DEFAULT_CONFIG),
+                credentialsJson,
+                getString(GCP_CREDENTIALS_PATH_CONFIG)
+            );
+        } catch (final IllegalArgumentException e) {
+            final String message = e.getMessage()
+                .replace("credentialsPath", GCP_CREDENTIALS_PATH_CONFIG)
+                .replace("credentialsJson", GCP_CREDENTIALS_JSON_CONFIG)
+                .replace("defaultCredentials", GCP_CREDENTIALS_DEFAULT_CONFIG);
+            throw new ConfigException(message);
         }
     }
 
@@ -148,7 +142,7 @@ Integer resumableUploadChunkSize() {
     }
 
     Credentials credentials() {
-        final boolean defaultCredentials = getBoolean(GCP_CREDENTIALS_DEFAULT_CONFIG);
+        final Boolean defaultCredentials = getBoolean(GCP_CREDENTIALS_DEFAULT_CONFIG);
         final Password credentialsJsonPwd = getPassword(GCP_CREDENTIALS_JSON_CONFIG);
         final String credentialsJson = credentialsJsonPwd == null ? null : credentialsJsonPwd.value();
         final String credentialsPath = getString(GCP_CREDENTIALS_PATH_CONFIG);

storage/gcs/src/test/java/io/aiven/kafka/tieredstorage/storage/gcs/CredentialsBuilderTest.java

@@ -18,13 +18,17 @@
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.stream.Stream;
 
 import com.google.auth.Credentials;
 import com.google.auth.oauth2.GoogleCredentials;
 import com.google.auth.oauth2.UserCredentials;
 import com.google.cloud.NoCredentials;
 import com.google.common.io.Resources;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 
@@ -33,24 +37,32 @@
 
 class CredentialsBuilderTest {
     @Test
-    void defaultAndJsonProvided() {
-        assertThatThrownBy(() -> CredentialsBuilder.build(true, "{}", null))
+    void allNull() {
+        assertThatThrownBy(() -> CredentialsBuilder.build(null, null, null))
             .isInstanceOf(IllegalArgumentException.class)
-            .hasMessage("defaultCredentials == true and credentialsJson != null cannot be simultaneously.");
+            .hasMessage("All defaultCredentials, credentialsJson, and credentialsPath cannot be null simultaneously.");
     }
 
-    @Test
-    void defaultAndPathProvided() {
-        assertThatThrownBy(() -> CredentialsBuilder.build(true, null, "file.json"))
+    @ParameterizedTest
+    @MethodSource("provideMoreThanOneNonNull")
+    void moreThanOneNonNull(final Boolean defaultCredentials,
+                            final String credentialsJson,
+                            final String credentialsPath) {
+        assertThatThrownBy(() -> CredentialsBuilder.build(defaultCredentials, credentialsJson, credentialsPath))
             .isInstanceOf(IllegalArgumentException.class)
-            .hasMessage("defaultCredentials == true and credentialsPath != null cannot be simultaneously.");
+            .hasMessage("Only one of defaultCredentials, credentialsJson, and credentialsPath can be non-null.");
     }
 
-    @Test
-    void bothCredentialsProvided() {
-        assertThatThrownBy(() -> CredentialsBuilder.build(false, "{}", "file.json"))
-            .isInstanceOf(IllegalArgumentException.class)
-            .hasMessage("Both credentialsPath and credentialsJson cannot be non-null.");
+    private static Stream<Arguments> provideMoreThanOneNonNull() {
+        return Stream.of(
+            Arguments.of(true, "json", "path"),
+            Arguments.of(false, "json", "path"),
+            Arguments.of(true, "json", null),
+            Arguments.of(false, "json", null),
+            Arguments.of(true, null, "path"),
+            Arguments.of(false, null, "path"),
+            Arguments.of(null, "json", "path")
+        );
     }
 
     @Test
@@ -75,7 +87,7 @@ void fromJson() throws IOException {
         final String credentialsJson = Resources.toString(
             Thread.currentThread().getContextClassLoader().getResource("test_gcs_credentials.json"),
             StandardCharsets.UTF_8);
-        final Credentials credentials = CredentialsBuilder.build(false, credentialsJson, null);
+        final Credentials credentials = CredentialsBuilder.build(null, credentialsJson, null);
         assertCredentials(credentials);
     }
 
@@ -85,7 +97,7 @@ void fromPath() throws IOException {
             .getContextClassLoader()
             .getResource("test_gcs_credentials.json")
             .getPath();
-        final Credentials credentials = CredentialsBuilder.build(false, null, credentialsPath);
+        final Credentials credentials = CredentialsBuilder.build(null, null, credentialsPath);
         assertCredentials(credentials);
     }
 

storage/gcs/src/test/java/io/aiven/kafka/tieredstorage/storage/gcs/GcsStorageConfigTest.java

@@ -16,13 +16,18 @@
 
 package io.aiven.kafka.tieredstorage.storage.gcs;
 
+import java.util.HashMap;
 import java.util.Map;
+import java.util.stream.Stream;
 
 import org.apache.kafka.common.config.ConfigException;
 
 import com.google.auth.oauth2.GoogleCredentials;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 import org.mockito.junit.jupiter.MockitoExtension;
@@ -35,7 +40,9 @@ class GcsStorageConfigTest {
     @Test
     void minimalConfig() {
         final String bucketName = "b1";
-        final Map<String, Object> configs = Map.of("gcs.bucket.name", bucketName);
+        final Map<String, Object> configs = Map.of(
+            "gcs.bucket.name", bucketName,
+            "gcs.credentials.default", "true");
         final GcsStorageConfig config = new GcsStorageConfig(configs);
         assertThat(config.bucketName()).isEqualTo(bucketName);
         assertThat(config.endpointUrl()).isNull();
@@ -99,44 +106,57 @@ void emptyPathCredentials() {
     }
 
     @Test
-    void jsonAndPathCredentialsSet() {
+    void allCredentialsNull() {
         final var props = Map.of(
-            "gcs.bucket.name", "bucket",
-            "gcs.credentials.default", "false",
-            "gcs.credentials.json", "json",
-            "gcs.credentials.path", "path");
+            "gcs.bucket.name", "bucket"
+        );
         assertThatThrownBy(() -> new GcsStorageConfig(props))
             .isInstanceOf(ConfigException.class)
-            .hasMessage("gcs.credentials.json and gcs.credentials.path cannot be set together");
+            .hasMessage("All gcs.credentials.default, gcs.credentials.json, and gcs.credentials.path "
+                + "cannot be null simultaneously.");
     }
 
-    @Test
-    void jsonAndDefaultCredentialsSet() {
-        final var props = Map.of(
-            "gcs.bucket.name", "bucket",
-            "gcs.credentials.json", "json",
-            "gcs.credentials.default", "true");
+    @ParameterizedTest
+    @MethodSource("provideMoreThanOneCredentialsNonNull")
+    void moreThanOneCredentialsNonNull(final Boolean defaultCredentials,
+                                       final String credentialsJson,
+                                       final String credentialsPath) {
+        final Map<String, Object> props = new HashMap<>();
+        props.put("gcs.bucket.name", "bucket");
+
+        if (defaultCredentials != null) {
+            props.put("gcs.credentials.default", Boolean.toString(defaultCredentials));
+        }
+        if (credentialsJson != null) {
+            props.put("gcs.credentials.json", credentialsJson);
+        }
+        if (credentialsPath != null) {
+            props.put("gcs.credentials.path", credentialsPath);
+        }
         assertThatThrownBy(() -> new GcsStorageConfig(props))
             .isInstanceOf(ConfigException.class)
-            .hasMessage("gcs.credentials.json and gcs.credentials.default cannot be set together");
+            .hasMessage("Only one of gcs.credentials.default, gcs.credentials.json, and gcs.credentials.path "
+                + "can be non-null.");
     }
 
-    @Test
-    void pathAndDefaultCredentialsSet() {
-        final var props = Map.of(
-            "gcs.bucket.name", "bucket",
-            "gcs.credentials.path", "path",
-            "gcs.credentials.default", "true");
-        assertThatThrownBy(() -> new GcsStorageConfig(props))
-            .isInstanceOf(ConfigException.class)
-            .hasMessage("gcs.credentials.path and gcs.credentials.default cannot be set together");
+    private static Stream<Arguments> provideMoreThanOneCredentialsNonNull() {
+        return Stream.of(
+            Arguments.of(true, "json", "path"),
+            Arguments.of(false, "json", "path"),
+            Arguments.of(true, "json", null),
+            Arguments.of(false, "json", null),
+            Arguments.of(true, null, "path"),
+            Arguments.of(false, null, "path"),
+            Arguments.of(null, "json", "path")
+        );
     }
 
     @Test
     void resumableUploadChunkSize() {
         final GcsStorageConfig config = new GcsStorageConfig(
             Map.of(
                 "gcs.bucket.name", "b1",
+                "gcs.credentials.default", "true",
                 "gcs.resumable.upload.chunk.size", Integer.toString(10 * 1024 * 1024)
             )
         );