diff --git a/CONFIGURATION.md b/CONFIGURATION.md index 8a7cd52c..60123315 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -49,7 +49,6 @@ These parameters fine-tune the low-level data streaming behavior. They allow you | `analytics-core.adaptive-read.sequential-read-threshold` | Threshold for number of sequential reads to switch to sequential mode. | `3` | | `analytics-core.random-read.min-request-size` | Minimum request size for random reads. If the requested read size is smaller, it reads up to this size. | `131072` (128 KB) | - ### Telemetry and Monitoring These settings enable the emission of deep internal metrics—such as cache hit rates, operational durations, and throughput—to local logging consoles or distributed OpenTelemetry backends like Google Cloud Monitoring. diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/FlatNamespaceStrategyImpl.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/FlatNamespaceStrategyImpl.java new file mode 100644 index 00000000..11c8b612 --- /dev/null +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/FlatNamespaceStrategyImpl.java @@ -0,0 +1,26 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed 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. + */ + +package com.google.cloud.gcs.analyticscore.client; + +final class FlatNamespaceStrategyImpl implements NamespaceStrategy { + + private final GcsClient gcsClient; + + FlatNamespaceStrategyImpl(GcsClient gcsClient) { + this.gcsClient = gcsClient; + } +} diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsClient.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsClient.java index 47d4d0ab..1b20984e 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsClient.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsClient.java @@ -43,6 +43,8 @@ WritableByteChannel createWriteChannel(GcsItemId itemId, GcsWriteOptions options /** Fetches object metadata. */ GcsItemInfo getGcsItemInfo(GcsItemId itemId) throws IOException; + boolean isHnsBucket(String bucketName) throws IOException; + /** Close the client. */ void close(); } diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsClientImpl.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsClientImpl.java index b9ee619c..07e7461f 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsClientImpl.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsClientImpl.java @@ -171,6 +171,11 @@ BucketProperties getBucketProperties(String bucketName) throws IOException { } } + @Override + public boolean isHnsBucket(String bucketName) throws IOException { + return getBucketProperties(bucketName).isHnsEnabled(); + } + @Override public void close() { try { diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemImpl.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemImpl.java index 322e68d1..f8aa5d7c 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemImpl.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemImpl.java @@ -34,6 +34,7 @@ import com.google.common.collect.ImmutableList; import com.google.common.util.concurrent.ThreadFactoryBuilder; import java.io.IOException; +import java.io.UncheckedIOException; import java.net.URI; import java.nio.channels.WritableByteChannel; import java.util.Collections; @@ -44,16 +45,44 @@ public class GcsFileSystemImpl implements GcsFileSystem { + /** + * Status or list calls (e.g., getting file info or listing a directory) block on I/O. A core pool + * size of 2 allows basic concurrency without significant resource overhead. + */ + private static final int CACHED_EXECUTOR_CORE_POOL_SIZE = 2; + + /** + * Using a 30-second keep-alive enables efficient thread reuse during intermittent spikes in + * status and list requests, while ensuring rapid resource cleanup during periods of inactivity. + */ + private static final int CACHED_EXECUTOR_KEEP_ALIVE_SECONDS = 30; + + /** + * Status and list calls block on I/O. An unbounded maximum pool size, combined with a + * zero-capacity SynchronousQueue, ensures tasks are never queued and new threads are immediately + * allocated to handle concurrent operations. + */ + private static final int CACHED_EXECUTOR_MAX_POOL_SIZE = Integer.MAX_VALUE; + + /** + * The maximum amount of time in seconds to wait for background thread pools to gracefully + * terminate upon file system closure. + */ + private static final int EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS = 10; + private final GcsClient gcsClient; private final GcsFileSystemOptions fileSystemOptions; - private final Supplier executorServiceSupplier; - + private final Supplier readExecutorServiceSupplier; + private final Supplier listExecutorServiceSupplier; private final Telemetry telemetry; private final AnalyticsCacheManager cacheManager; + private final FlatNamespaceStrategyImpl flatStrategy; + private final HierarchicalNamespaceStrategyImpl hnsStrategy; public GcsFileSystemImpl(GcsFileSystemOptions fileSystemOptions) { this.fileSystemOptions = fileSystemOptions; - this.executorServiceSupplier = initializeExecutionServiceSupplier(); + this.readExecutorServiceSupplier = initializeReadExecutionServiceSupplier(); + this.listExecutorServiceSupplier = initializeListExecutionServiceSupplier(); this.telemetry = createTelemetry(fileSystemOptions.getAnalyticsCoreTelemetryOptions()); this.cacheManager = new AnalyticsCacheManager(fileSystemOptions.getGcsCacheOptions()); this.gcsClient = @@ -63,12 +92,17 @@ public GcsFileSystemImpl(GcsFileSystemOptions fileSystemOptions) { Collections.emptyMap(), recorder -> new GcsClientImpl( - fileSystemOptions.getGcsClientOptions(), executorServiceSupplier, telemetry)); + fileSystemOptions.getGcsClientOptions(), + readExecutorServiceSupplier, + telemetry)); + this.flatStrategy = new FlatNamespaceStrategyImpl(this.gcsClient); + this.hnsStrategy = new HierarchicalNamespaceStrategyImpl(this.gcsClient); } public GcsFileSystemImpl(Credentials credentials, GcsFileSystemOptions fileSystemOptions) { this.fileSystemOptions = fileSystemOptions; - this.executorServiceSupplier = initializeExecutionServiceSupplier(); + this.readExecutorServiceSupplier = initializeReadExecutionServiceSupplier(); + this.listExecutorServiceSupplier = initializeListExecutionServiceSupplier(); this.telemetry = createTelemetry(fileSystemOptions.getAnalyticsCoreTelemetryOptions()); this.cacheManager = new AnalyticsCacheManager(fileSystemOptions.getGcsCacheOptions()); this.gcsClient = @@ -80,8 +114,10 @@ public GcsFileSystemImpl(Credentials credentials, GcsFileSystemOptions fileSyste new GcsClientImpl( credentials, fileSystemOptions.getGcsClientOptions(), - executorServiceSupplier, + readExecutorServiceSupplier, telemetry)); + this.flatStrategy = new FlatNamespaceStrategyImpl(this.gcsClient); + this.hnsStrategy = new HierarchicalNamespaceStrategyImpl(this.gcsClient); } @VisibleForTesting @@ -101,9 +137,36 @@ public GcsFileSystemImpl(Credentials credentials, GcsFileSystemOptions fileSyste AnalyticsCacheManager cacheManager) { this.gcsClient = gcsClient; this.fileSystemOptions = fileSystemOptions; - this.executorServiceSupplier = initializeExecutionServiceSupplier(); + this.readExecutorServiceSupplier = initializeReadExecutionServiceSupplier(); + this.listExecutorServiceSupplier = initializeListExecutionServiceSupplier(); this.telemetry = telemetry; this.cacheManager = cacheManager; + this.flatStrategy = new FlatNamespaceStrategyImpl(this.gcsClient); + this.hnsStrategy = new HierarchicalNamespaceStrategyImpl(this.gcsClient); + } + + @VisibleForTesting + NamespaceStrategy resolveStrategy(String bucketName) throws IOException { + checkNotNull(bucketName, "bucketName cannot be null"); + if (!fileSystemOptions.isHnsApiEnabled()) { + return flatStrategy; + } + + BucketProperties properties = + cacheManager.getBucketProperties( + bucketName, + name -> { + try { + return BucketProperties.create(gcsClient.isHnsBucket(name)); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + }); + + if (properties.isHnsEnabled()) { + return hnsStrategy; + } + return flatStrategy; } @Override @@ -163,16 +226,39 @@ public AnalyticsCacheManager getCacheManager() { return cacheManager; } + @VisibleForTesting + FlatNamespaceStrategyImpl getFlatStrategy() { + return flatStrategy; + } + + @VisibleForTesting + HierarchicalNamespaceStrategyImpl getHnsStrategy() { + return hnsStrategy; + } + @Override public void close() { - ExecutorService executorService = executorServiceSupplier.get(); - executorService.shutdown(); + ExecutorService readExecutorService = readExecutorServiceSupplier.get(); + ExecutorService listExecutorService = listExecutorServiceSupplier.get(); + readExecutorService.shutdown(); + listExecutorService.shutdown(); try { - if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) { - executorService.shutdownNow(); + // Wait a total of LIST_EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS for both thread pools to terminate. + long deadline = + System.nanoTime() + TimeUnit.SECONDS.toNanos(EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS); + // First, wait for the read executor service to terminate. + if (!readExecutorService.awaitTermination( + EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + readExecutorService.shutdownNow(); + } + // Then, wait for the cached executor service to terminate, with the remaining time. + if (!listExecutorService.awaitTermination( + Math.max(0, deadline - System.nanoTime()), TimeUnit.NANOSECONDS)) { + listExecutorService.shutdownNow(); } } catch (InterruptedException e) { - executorService.shutdownNow(); + readExecutorService.shutdownNow(); + listExecutorService.shutdownNow(); Thread.currentThread().interrupt(); } gcsClient.close(); @@ -206,7 +292,7 @@ static Telemetry createTelemetry(TelemetryOptions telemetryOptions) { } @VisibleForTesting - Supplier initializeExecutionServiceSupplier() { + Supplier initializeReadExecutionServiceSupplier() { return Suppliers.memoize( () -> new ThreadPoolExecutor( @@ -220,4 +306,33 @@ Supplier initializeExecutionServiceSupplier() { .setDaemon(true) .build())); } + + @VisibleForTesting + Supplier initializeListExecutionServiceSupplier() { + return Suppliers.memoize( + () -> { + if (fileSystemOptions.isListParallelEnabled()) { + return createCachedExecutor(); + } + return new LazyExecutorService(); + }); + } + + private static ExecutorService createCachedExecutor() { + ThreadPoolExecutor service = + new ThreadPoolExecutor( + /* corePoolSize= */ CACHED_EXECUTOR_CORE_POOL_SIZE, + /* maximumPoolSize= */ CACHED_EXECUTOR_MAX_POOL_SIZE, + /* keepAliveTime= */ CACHED_EXECUTOR_KEEP_ALIVE_SECONDS, + TimeUnit.SECONDS, + new java.util.concurrent.SynchronousQueue<>(), + new ThreadFactoryBuilder() + .setNameFormat("gcs-filesystem-cached-pool-%d") + .setDaemon(true) + .build()); + // allowCoreThreadTimeOut needs to be enabled for cases where the encapsulating class does not + // properly shut down the executor, preventing thread leaks. + service.allowCoreThreadTimeOut(true); + return service; + } } diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemOptions.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemOptions.java index 95ab244c..31456e7b 100644 --- a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemOptions.java +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemOptions.java @@ -25,6 +25,8 @@ public abstract class GcsFileSystemOptions { private static final String READ_THREAD_COUNT_KEY = "analytics-core.read.thread.count"; private static final String CLIENT_TYPE_KEY = "client.type"; + private static final String HNS_API_ENABLED_KEY = "analytics-core.hierarchical.namespace.enable"; + private static final String LIST_PARALLEL_ENABLED_KEY = "analytics-core.list.parallel.enabled"; /** Cloud Storage client to use. */ public enum ClientType { @@ -43,12 +45,18 @@ public enum ClientType { public abstract TelemetryOptions getAnalyticsCoreTelemetryOptions(); + public abstract boolean isHnsApiEnabled(); + + public abstract boolean isListParallelEnabled(); + public abstract Builder toBuilder(); public static Builder builder() { return new AutoValue_GcsFileSystemOptions.Builder() .setReadThreadCount(16) .setClientType(ClientType.HTTP_CLIENT) + .setHnsApiEnabled(true) + .setListParallelEnabled(true) .setGcsClientOptions(GcsClientOptions.builder().build()) .setGcsCacheOptions(GcsCacheOptions.builder().build()) .setAnalyticsCoreTelemetryOptions(TelemetryOptions.builder().build()); @@ -65,6 +73,15 @@ public static GcsFileSystemOptions createFromOptions( optionsBuilder.setClientType( ClientType.valueOf(analyticsCoreOptions.get(prefix + CLIENT_TYPE_KEY))); } + if (analyticsCoreOptions.containsKey(prefix + HNS_API_ENABLED_KEY)) { + optionsBuilder.setHnsApiEnabled( + Boolean.parseBoolean(analyticsCoreOptions.get(prefix + HNS_API_ENABLED_KEY))); + } + if (analyticsCoreOptions.containsKey(prefix + LIST_PARALLEL_ENABLED_KEY)) { + optionsBuilder.setListParallelEnabled( + Boolean.parseBoolean(analyticsCoreOptions.get(prefix + LIST_PARALLEL_ENABLED_KEY))); + } + optionsBuilder.setGcsClientOptions( GcsClientOptions.createFromOptions(analyticsCoreOptions, prefix)); optionsBuilder.setGcsCacheOptions( @@ -84,6 +101,10 @@ public abstract static class Builder { public abstract Builder setReadThreadCount(int readThreadCount); + public abstract Builder setHnsApiEnabled(boolean isHnsApiEnabled); + + public abstract Builder setListParallelEnabled(boolean isListParallelEnabled); + public abstract Builder setGcsClientOptions(GcsClientOptions gcsClientOptions); /** Sets the configuration options for the GCS caching layer. */ diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/HierarchicalNamespaceStrategyImpl.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/HierarchicalNamespaceStrategyImpl.java new file mode 100644 index 00000000..b05e8652 --- /dev/null +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/HierarchicalNamespaceStrategyImpl.java @@ -0,0 +1,25 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed 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. + */ + +package com.google.cloud.gcs.analyticscore.client; + +final class HierarchicalNamespaceStrategyImpl implements NamespaceStrategy { + private final GcsClient gcsClient; + + HierarchicalNamespaceStrategyImpl(GcsClient gcsClient) { + this.gcsClient = gcsClient; + } +} diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/LazyExecutorService.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/LazyExecutorService.java new file mode 100644 index 00000000..09c01072 --- /dev/null +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/LazyExecutorService.java @@ -0,0 +1,179 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed 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. + */ + +package com.google.cloud.gcs.analyticscore.client; + +import java.util.Collections; +import java.util.List; +import java.util.concurrent.AbstractExecutorService; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.RunnableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +/** + * A lightweight, lazy ExecutorService that defers task execution until {@code Future.get()} is + * called. Execution happens synchronously on the thread that invokes {@code get()}. + * + *

A returned Future represents a pending task. Upon the first invocation of its {@code get()} + * method, the task executes and its result is permanently cached. + * + *

Timeout Limitation: Because execution is strictly synchronous on the caller's thread, true + * preemptive timeouts are not supported. If a timeout is provided to {@code get()}, it only checks + * if the timeout has already elapsed prior to execution. If execution begins, it will block + * indefinitely until the task completes, ignoring the timeout duration during execution. + * + *

Both this class and the returned Future are thread-safe. + */ +final class LazyExecutorService extends AbstractExecutorService { + + private volatile boolean isShutdown = false; + + @Override + public void shutdown() { + isShutdown = true; + } + + @Override + public List shutdownNow() { + isShutdown = true; + return Collections.emptyList(); + } + + @Override + public boolean isShutdown() { + return isShutdown; + } + + /** Returns true if the executor has been shut down, since there are no asynchronous tasks. */ + @Override + public boolean isTerminated() { + return isShutdown; + } + + /** Returns whether the executor has been shut down. */ + @Override + public boolean awaitTermination(long timeout, TimeUnit unit) { + return isShutdown; + } + + @Override + public void execute(Runnable command) { + throw new RejectedExecutionException("Use submit instead of execute."); + } + + /** + * Bulk execution operations (invokeAll, invokeAny) are not supported by this lazy executor. Tasks + * must be explicitly submitted and resolved individually via their returned Futures. + */ + @Override + public List> invokeAll(java.util.Collection> tasks) { + throw new UnsupportedOperationException("LazyExecutorService does not support invokeAll"); + } + + @Override + public List> invokeAll( + java.util.Collection> tasks, long timeout, TimeUnit unit) { + throw new UnsupportedOperationException("LazyExecutorService does not support invokeAll"); + } + + @Override + public T invokeAny(java.util.Collection> tasks) { + throw new UnsupportedOperationException("LazyExecutorService does not support invokeAny"); + } + + @Override + public T invokeAny( + java.util.Collection> tasks, long timeout, TimeUnit unit) { + throw new UnsupportedOperationException("LazyExecutorService does not support invokeAny"); + } + + @Override + public Future submit(Runnable task) { + return submit(Executors.callable(task)); + } + + @Override + public Future submit(Runnable task, T result) { + return submit(Executors.callable(task, result)); + } + + @Override + public Future submit(Callable task) { + if (task == null) throw new NullPointerException(); + if (isShutdown) { + throw new RejectedExecutionException("Executor is shut down"); + } + return newTaskFor(task); + } + + private final class LazyFutureTask extends FutureTask { + LazyFutureTask(Callable callable) { + super(callable); + } + + @Override + public V get() throws InterruptedException, ExecutionException { + if (!isDone()) { + if (Thread.interrupted()) { + throw new InterruptedException(); + } + if (isShutdown) { + cancel(false); + } else { + run(); // Execute on the caller's thread when get() is called + } + } + return super.get(); + } + + /** + * Note: Because this implementation executes the task synchronously on the calling thread, the + * provided timeout is inherently ignored during the actual execution of the task. The calling + * thread will remain blocked until {@code run()} completes, at which point the timeout logic + * evaluates. True preemptive timeouts are not supported. + */ + @Override + public V get(long timeout, TimeUnit unit) + throws InterruptedException, ExecutionException, TimeoutException { + if (!isDone()) { + if (Thread.interrupted()) { + throw new InterruptedException(); + } + if (timeout <= 0) { + throw new TimeoutException(); + } + if (isShutdown) { + cancel(false); + } else { + run(); + } + } + return super.get(timeout, unit); + } + } + + /** Wraps the given callable into a LazyFutureTask that overrides get() to execute the task. */ + @Override + protected RunnableFuture newTaskFor(Callable callable) { + return new LazyFutureTask<>(callable); + } +} diff --git a/client/src/main/java/com/google/cloud/gcs/analyticscore/client/NamespaceStrategy.java b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/NamespaceStrategy.java new file mode 100644 index 00000000..9f896b7f --- /dev/null +++ b/client/src/main/java/com/google/cloud/gcs/analyticscore/client/NamespaceStrategy.java @@ -0,0 +1,34 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed 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. + */ + +package com.google.cloud.gcs.analyticscore.client; + +/** + * Strategy interface for handling directory operations across different namespace models (Flat vs. + * HNS). + * + *

Methods for directory operations will be added in follow-up PRs. These methods will include: + * + *

    + *
  • {@code GcsItemInfo getFileInfo(GcsItemId id, PathType pathType) throws IOException;} + *
  • {@code void createDirectory(GcsItemId id) throws IOException;} + *
  • {@code boolean isDirectoryEmpty(GcsItemId id) throws IOException;} + *
  • {@code void renameDirectory(GcsItemId src, GcsItemId dst) throws IOException;} + *
  • {@code java.util.List listObjectInfo(GcsItemId id) throws IOException;} + *
  • {@code java.util.List listRecursive(GcsItemId id) throws IOException;} + *
+ */ +interface NamespaceStrategy {} diff --git a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsClientImplTest.java b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsClientImplTest.java index fc0598d9..0a229db7 100644 --- a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsClientImplTest.java +++ b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsClientImplTest.java @@ -83,8 +83,10 @@ class GcsClientImplTest { private static final String TEST_OBJECT_ID = "test-object-id"; private static final String TEST_WRITE_OBJECT = "test-write-object"; private static final String TEST_NULL_OPTIONS_OBJECT = "test-null-options"; - private static final String TEST_NON_EXISTENT_OBJECT = "non-existent"; - private static final String NON_EXISTENT_BUCKET = "non-existent-bucket"; + private static final String TEST_NON_EXISTENT_OBJECT = "non-existent-object"; + private static final String TEST_HNS_BUCKET = "hns-bucket"; + private static final String TEST_FLAT_BUCKET = "flat-bucket"; + private static final String TEST_NON_EXISTENT_BUCKET = "non-existent-bucket"; private static final String TEST_OBJECT_NAME = "test-object-name"; private static final String BLOB_WRITE_SESSION_CONFIG_FIELD = "blobWriteSessionConfig"; private static final int MB = 1024 * 1024; @@ -310,39 +312,61 @@ void getBucketProperties_nullBucketName_throwsNullPointerException() { } @Test - void getBucketProperties_hnsEnabled_returnsTrue() throws IOException { + void getBucketProperties_hnsBucket_returnsTrue() throws IOException { Storage mockStorage = mock(Storage.class); GcsClientImpl localGcsClient = createClientWithMockStorage(mockStorage); Bucket mockBucket = mockBucketWithHns(true); - doReturn(mockBucket).when(mockStorage).get(eq("hns-bucket"), any(BucketGetOption.class)); + doReturn(mockBucket).when(mockStorage).get(eq(TEST_HNS_BUCKET), any(BucketGetOption.class)); - BucketProperties bucketProperties = localGcsClient.getBucketProperties("hns-bucket"); + BucketProperties bucketProperties = localGcsClient.getBucketProperties(TEST_HNS_BUCKET); assertThat(bucketProperties.isHnsEnabled()).isTrue(); } @Test - void getBucketProperties_hnsDisabled_returnsFalse() throws IOException { + void isHnsBucket_hnsBucket_returnsTrue() throws IOException { + Storage mockStorage = mock(Storage.class); + GcsClientImpl localGcsClient = createClientWithMockStorage(mockStorage); + Bucket mockBucket = mockBucketWithHns(true); + doReturn(mockBucket).when(mockStorage).get(eq(TEST_HNS_BUCKET), any(BucketGetOption.class)); + + boolean isHns = localGcsClient.isHnsBucket(TEST_HNS_BUCKET); + + assertThat(isHns).isTrue(); + } + + @Test + void getBucketProperties_flatBucket_returnsFalse() throws IOException { Storage mockStorage = mock(Storage.class); GcsClientImpl localGcsClient = createClientWithMockStorage(mockStorage); Bucket mockBucket = mockBucketWithHns(false); - doReturn(mockBucket).when(mockStorage).get(eq("flat-bucket"), any(BucketGetOption.class)); + doReturn(mockBucket).when(mockStorage).get(eq(TEST_FLAT_BUCKET), any(BucketGetOption.class)); - BucketProperties bucketProperties = localGcsClient.getBucketProperties("flat-bucket"); + BucketProperties bucketProperties = localGcsClient.getBucketProperties(TEST_FLAT_BUCKET); assertThat(bucketProperties.isHnsEnabled()).isFalse(); } @Test - void getBucketProperties_hnsNull_returnsFalse() throws IOException { + void isHnsBucket_flatBucket_returnsFalse() throws IOException { + Storage mockStorage = mock(Storage.class); + GcsClientImpl localGcsClient = createClientWithMockStorage(mockStorage); + Bucket mockBucket = mockBucketWithHns(false); + doReturn(mockBucket).when(mockStorage).get(eq(TEST_FLAT_BUCKET), any(BucketGetOption.class)); + + boolean isHns = localGcsClient.isHnsBucket(TEST_FLAT_BUCKET); + + assertThat(isHns).isFalse(); + } + + @Test + void getBucketProperties_missingHnsProperty_returnsFalse() throws IOException { Storage mockStorage = mock(Storage.class); GcsClientImpl localGcsClient = createClientWithMockStorage(mockStorage); Bucket mockBucket = mockBucketWithHns(null); - doReturn(mockBucket) - .when(mockStorage) - .get(eq("flat-bucket-null-hns"), any(BucketGetOption.class)); + doReturn(mockBucket).when(mockStorage).get(eq(TEST_BUCKET), any(BucketGetOption.class)); - BucketProperties bucketProperties = localGcsClient.getBucketProperties("flat-bucket-null-hns"); + BucketProperties bucketProperties = localGcsClient.getBucketProperties(TEST_BUCKET); assertThat(bucketProperties.isHnsEnabled()).isFalse(); } @@ -351,9 +375,9 @@ void getBucketProperties_hnsNull_returnsFalse() throws IOException { void getBucketProperties_bucketNotFound_returnsDisabledHns() throws Exception { Storage mockStorage = mock(Storage.class); GcsClientImpl localGcsClient = createClientWithMockStorage(mockStorage); - doReturn(null).when(mockStorage).get(eq(NON_EXISTENT_BUCKET), any(BucketGetOption.class)); + doReturn(null).when(mockStorage).get(eq(TEST_NON_EXISTENT_BUCKET), any(BucketGetOption.class)); - BucketProperties properties = localGcsClient.getBucketProperties(NON_EXISTENT_BUCKET); + BucketProperties properties = localGcsClient.getBucketProperties(TEST_NON_EXISTENT_BUCKET); assertThat(properties.isHnsEnabled()).isFalse(); } @@ -505,9 +529,12 @@ void create_whenBucketOrObjectNotFound_throwsFileNotFoundException() throws Exce Storage mockStorage = mock(Storage.class); GcsClientImpl clientWithMock = createClientWithMockStorage(mockStorage); GcsItemId itemId = - GcsItemId.builder().setBucketName(NON_EXISTENT_BUCKET).setObjectName(TEST_OBJECT).build(); + GcsItemId.builder() + .setBucketName(TEST_NON_EXISTENT_BUCKET) + .setObjectName(TEST_NON_EXISTENT_OBJECT) + .build(); BlobInfo blobInfo = - BlobInfo.newBuilder(BlobId.of(NON_EXISTENT_BUCKET, TEST_OBJECT)) + BlobInfo.newBuilder(BlobId.of(TEST_NON_EXISTENT_BUCKET, TEST_NON_EXISTENT_OBJECT)) .setContentType("application/octet-stream") .build(); StorageException e404 = new StorageException(404, "Not Found"); diff --git a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemImplTest.java b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemImplTest.java index 3897c3e9..b6bfcf7a 100644 --- a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemImplTest.java +++ b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemImplTest.java @@ -35,6 +35,7 @@ import com.google.common.base.Supplier; import com.google.common.collect.ImmutableList; import java.io.IOException; +import java.io.UncheckedIOException; import java.net.URI; import java.net.URISyntaxException; import java.nio.channels.WritableByteChannel; @@ -66,6 +67,7 @@ class GcsFileSystemImplTest { GcsFileSystemOptions.builder().setGcsClientOptions(TEST_GCS_CLIENT_OPTIONS).build(); @Mock private GcsClient mockClient; + private GcsFileSystem gcsFileSystem; @BeforeEach @@ -88,6 +90,10 @@ void constructor_withCredentials_createsClientWithProvidedCredentials() { assertThat(gcsClientImpl.storage.getOptions().getCredentials()) .isEqualTo(NoCredentials.getInstance()); + assertThat(gcsFileSystem.getTelemetry()).isNotNull(); + assertThat(gcsFileSystem.getCacheManager()).isNotNull(); + assertThat(gcsFileSystem.getFlatStrategy()).isNotNull(); + assertThat(gcsFileSystem.getHnsStrategy()).isNotNull(); } } @@ -104,12 +110,17 @@ void constructor_withFileSystemOptions_createsClientWithDefaultCredentials() { assertThat(gcsFileSystem.getFileSystemOptions()).isSameInstanceAs(fileSystemOptions); assertThat(gcsClient).isNotNull(); assertThat(gcsClient.storage.getOptions().getProjectId()).isEqualTo("test-project-default"); + assertThat(gcsFileSystem.getTelemetry()).isNotNull(); + assertThat(gcsFileSystem.getCacheManager()).isNotNull(); + assertThat(gcsFileSystem.getFlatStrategy()).isNotNull(); + assertThat(gcsFileSystem.getHnsStrategy()).isNotNull(); } } @Test - void constructor_withValidOptions_passesExecutorToClient() { + void constructor_withValidOptions_passesMemorizedExecutorServiceAndTelemetryToGcsClient() { final AtomicReference> capturedSupplier = new AtomicReference<>(); + final AtomicReference capturedTelemetry = new AtomicReference<>(); try (MockedConstruction mockGcsClientConstruction = Mockito.mockConstruction( GcsClientImpl.class, @@ -118,6 +129,9 @@ void constructor_withValidOptions_passesExecutorToClient() { Supplier supplier = (Supplier) context.arguments().get(1); capturedSupplier.set(supplier); + + Telemetry telemetry = (Telemetry) context.arguments().get(2); + capturedTelemetry.set(telemetry); })) { try (GcsFileSystemImpl fs = new GcsFileSystemImpl(TEST_GCS_FILESYSTEM_OPTIONS)) { @@ -128,6 +142,10 @@ void constructor_withValidOptions_passesExecutorToClient() { assertThat(capturedSupplier.get()).isNotNull(); assertThat(capturedSupplier.get().get()).isNotNull(); assertThat(executorService1).isEqualTo(executorService2); + assertThat(capturedTelemetry.get()).isNotNull(); + assertThat(capturedTelemetry.get()).isSameInstanceAs(fs.getTelemetry()); + assertThat(fs.getFlatStrategy()).isNotNull(); + assertThat(fs.getHnsStrategy()).isNotNull(); } } } @@ -298,80 +316,145 @@ void getFileInfo_withNonExistentItemId_shouldThrowException() throws IOException } @Test - void initializeExecutionServiceSupplier_shouldReturnMemoizedExecutorService() { + void initializeReadExecutionServiceSupplier_shouldReturnMemoizedExecutorService() { GcsFileSystemImpl fileSystemImpl = (GcsFileSystemImpl) gcsFileSystem; - Supplier executorServiceSupplier = - fileSystemImpl.initializeExecutionServiceSupplier(); + Supplier readExecutorServiceSupplier = + fileSystemImpl.initializeReadExecutionServiceSupplier(); - assertThat(executorServiceSupplier).isNotNull(); - assertThat(executorServiceSupplier.get()).isNotNull(); - assertThat(executorServiceSupplier.get()).isInstanceOf(ThreadPoolExecutor.class); - assertThat(((ThreadPoolExecutor) executorServiceSupplier.get()).getCorePoolSize()) + assertThat(readExecutorServiceSupplier).isNotNull(); + assertThat(readExecutorServiceSupplier.get()).isNotNull(); + assertThat(readExecutorServiceSupplier.get()).isInstanceOf(ThreadPoolExecutor.class); + assertThat(((ThreadPoolExecutor) readExecutorServiceSupplier.get()).getCorePoolSize()) .isEqualTo(16); + assertThat(readExecutorServiceSupplier.get()) + .isSameInstanceAs(readExecutorServiceSupplier.get()); + } + + @Test + void + initializeListExecutionServiceSupplier_whenListParallelEnabled_shouldReturnCachedExecutorService() { + GcsFileSystemImpl fileSystemImpl = (GcsFileSystemImpl) gcsFileSystem; + + Supplier listExecutorServiceSupplier = + fileSystemImpl.initializeListExecutionServiceSupplier(); + + assertThat(listExecutorServiceSupplier).isNotNull(); + assertThat(listExecutorServiceSupplier.get()).isNotNull(); + assertThat(listExecutorServiceSupplier.get()).isInstanceOf(ThreadPoolExecutor.class); + assertThat(listExecutorServiceSupplier.get()) + .isSameInstanceAs(listExecutorServiceSupplier.get()); + } + + @Test + void + initializeListExecutionServiceSupplier_whenListParallelDisabled_shouldReturnLazyExecutorService() + throws Exception { + GcsFileSystemOptions options = + TEST_GCS_FILESYSTEM_OPTIONS.toBuilder().setListParallelEnabled(false).build(); + // Use try-with-resources to ensure GcsFileSystemImpl is closed with its internal executors. + try (GcsFileSystemImpl fileSystemImpl = new GcsFileSystemImpl(mock(GcsClient.class), options)) { + + Supplier listExecutorServiceSupplier = + fileSystemImpl.initializeListExecutionServiceSupplier(); + + assertThat(listExecutorServiceSupplier).isNotNull(); + assertThat(listExecutorServiceSupplier.get()).isNotNull(); + assertThat(listExecutorServiceSupplier.get()).isInstanceOf(LazyExecutorService.class); + } } @Test void close_whenTerminationSucceeds_shutsDownGracefully() throws InterruptedException { - ExecutorService mockExecutorService = mock(ExecutorService.class); - when(mockExecutorService.awaitTermination(anyLong(), any(TimeUnit.class))).thenReturn(true); + ExecutorService mockReadExecutorService = mock(ExecutorService.class); + ExecutorService mockListExecutorService = mock(ExecutorService.class); + when(mockReadExecutorService.awaitTermination(anyLong(), any(TimeUnit.class))).thenReturn(true); + when(mockListExecutorService.awaitTermination(anyLong(), any(TimeUnit.class))).thenReturn(true); GcsFileSystemImpl fileSystemWithMockExecutor = new GcsFileSystemImpl(mockClient, TEST_GCS_FILESYSTEM_OPTIONS) { @Override - Supplier initializeExecutionServiceSupplier() { - return () -> mockExecutorService; + Supplier initializeReadExecutionServiceSupplier() { + return () -> mockReadExecutorService; + } + + @Override + Supplier initializeListExecutionServiceSupplier() { + return () -> mockListExecutorService; } }; fileSystemWithMockExecutor.close(); - InOrder inOrder = inOrder(mockExecutorService, mockClient); + InOrder inOrder = inOrder(mockReadExecutorService, mockListExecutorService, mockClient); - inOrder.verify(mockExecutorService).shutdown(); - inOrder.verify(mockExecutorService).awaitTermination(anyLong(), any(TimeUnit.class)); + inOrder.verify(mockReadExecutorService).shutdown(); + inOrder.verify(mockListExecutorService).shutdown(); + inOrder.verify(mockReadExecutorService).awaitTermination(anyLong(), any(TimeUnit.class)); + inOrder.verify(mockListExecutorService).awaitTermination(anyLong(), any(TimeUnit.class)); inOrder.verify(mockClient).close(); - verify(mockExecutorService, never()).shutdownNow(); + verify(mockReadExecutorService, never()).shutdownNow(); + verify(mockListExecutorService, never()).shutdownNow(); } @Test void close_whenTerminationTimesOut_shutsDownNow() throws InterruptedException { - ExecutorService mockExecutorService = mock(ExecutorService.class); - when(mockExecutorService.awaitTermination(anyLong(), any(TimeUnit.class))).thenReturn(false); + ExecutorService mockReadExecutorService = mock(ExecutorService.class); + ExecutorService mockListExecutorService = mock(ExecutorService.class); + when(mockReadExecutorService.awaitTermination(anyLong(), any(TimeUnit.class))) + .thenReturn(false); + when(mockListExecutorService.awaitTermination(anyLong(), any(TimeUnit.class))) + .thenReturn(false); GcsFileSystemImpl fileSystemWithMockExecutor = new GcsFileSystemImpl(mockClient, TEST_GCS_FILESYSTEM_OPTIONS) { @Override - Supplier initializeExecutionServiceSupplier() { - return () -> mockExecutorService; + Supplier initializeReadExecutionServiceSupplier() { + return () -> mockReadExecutorService; + } + + @Override + Supplier initializeListExecutionServiceSupplier() { + return () -> mockListExecutorService; } }; fileSystemWithMockExecutor.close(); - InOrder inOrder = inOrder(mockExecutorService, mockClient); - - inOrder.verify(mockExecutorService).shutdown(); - inOrder.verify(mockExecutorService).awaitTermination(anyLong(), any(TimeUnit.class)); - inOrder.verify(mockExecutorService).shutdownNow(); + InOrder inOrder = inOrder(mockReadExecutorService, mockListExecutorService, mockClient); + + inOrder.verify(mockReadExecutorService).shutdown(); + inOrder.verify(mockListExecutorService).shutdown(); + inOrder.verify(mockReadExecutorService).awaitTermination(anyLong(), any(TimeUnit.class)); + inOrder.verify(mockReadExecutorService).shutdownNow(); + inOrder.verify(mockListExecutorService).awaitTermination(anyLong(), any(TimeUnit.class)); + inOrder.verify(mockListExecutorService).shutdownNow(); inOrder.verify(mockClient).close(); } @Test void close_whenInterrupted_reInterruptsThreadAndShutsDownNow() throws InterruptedException { - ExecutorService mockExecutorService = mock(ExecutorService.class); - when(mockExecutorService.awaitTermination(anyLong(), any(TimeUnit.class))) + ExecutorService mockReadExecutorService = mock(ExecutorService.class); + ExecutorService mockListExecutorService = mock(ExecutorService.class); + when(mockReadExecutorService.awaitTermination(anyLong(), any(TimeUnit.class))) .thenThrow(new InterruptedException()); GcsFileSystemImpl fileSystemWithMockExecutor = new GcsFileSystemImpl(mockClient, TEST_GCS_FILESYSTEM_OPTIONS) { @Override - Supplier initializeExecutionServiceSupplier() { - return () -> mockExecutorService; + Supplier initializeReadExecutionServiceSupplier() { + return () -> mockReadExecutorService; + } + + @Override + Supplier initializeListExecutionServiceSupplier() { + return () -> mockListExecutorService; } }; fileSystemWithMockExecutor.close(); - InOrder inOrder = inOrder(mockExecutorService, mockClient); + InOrder inOrder = inOrder(mockReadExecutorService, mockListExecutorService, mockClient); - inOrder.verify(mockExecutorService).shutdown(); - inOrder.verify(mockExecutorService).awaitTermination(anyLong(), any(TimeUnit.class)); - inOrder.verify(mockExecutorService).shutdownNow(); + inOrder.verify(mockReadExecutorService).shutdown(); + inOrder.verify(mockListExecutorService).shutdown(); + inOrder.verify(mockReadExecutorService).awaitTermination(anyLong(), any(TimeUnit.class)); + inOrder.verify(mockReadExecutorService).shutdownNow(); + inOrder.verify(mockListExecutorService).shutdownNow(); inOrder.verify(mockClient).close(); assertThat(Thread.currentThread().isInterrupted()).isTrue(); Thread.interrupted(); // Clear interrupted status to not affect other tests @@ -574,6 +657,74 @@ void create_nullWriteOptions_delegatesToClientWithNullOptions() throws IOExcepti assertThat(resultChannel).isSameInstanceAs(mockChannel); } + @Test + void resolveStrategy_hnsFlagEnabledAndHnsBucket_returnsHnsStrategy() throws IOException { + GcsFileSystemOptions options = + GcsFileSystemOptions.builder() + .setGcsClientOptions(TEST_GCS_CLIENT_OPTIONS) + .setHnsApiEnabled(true) + .build(); + when(mockClient.isHnsBucket(TEST_BUCKET)).thenReturn(true); + + try (GcsFileSystemImpl gcsFileSystem = new GcsFileSystemImpl(mockClient, options)) { + NamespaceStrategy strategy = gcsFileSystem.resolveStrategy(TEST_BUCKET); + + assertThat(strategy).isInstanceOf(HierarchicalNamespaceStrategyImpl.class); + } + } + + @Test + void resolveStrategy_hnsFlagDisabled_returnsFlatStrategy() throws IOException { + GcsFileSystemOptions options = + GcsFileSystemOptions.builder() + .setGcsClientOptions(TEST_GCS_CLIENT_OPTIONS) + .setHnsApiEnabled(false) + .build(); + + try (GcsFileSystemImpl gcsFileSystem = new GcsFileSystemImpl(mockClient, options)) { + NamespaceStrategy strategy = gcsFileSystem.resolveStrategy(TEST_BUCKET); + + assertThat(strategy).isInstanceOf(FlatNamespaceStrategyImpl.class); + verify(mockClient, never()).isHnsBucket(anyString()); + } + } + + @Test + void resolveStrategy_hnsFlagEnabledAndFlatBucket_returnsFlatStrategy() throws IOException { + GcsFileSystemOptions options = + GcsFileSystemOptions.builder() + .setGcsClientOptions(TEST_GCS_CLIENT_OPTIONS) + .setHnsApiEnabled(true) + .build(); + when(mockClient.isHnsBucket(TEST_BUCKET)).thenReturn(false); + + try (GcsFileSystemImpl gcsFileSystem = new GcsFileSystemImpl(mockClient, options)) { + NamespaceStrategy strategy = gcsFileSystem.resolveStrategy(TEST_BUCKET); + + assertThat(strategy).isInstanceOf(FlatNamespaceStrategyImpl.class); + } + } + + @Test + void resolveStrategy_isHnsBucketThrowsIoException_throwsUncheckedIOException() + throws IOException { + GcsFileSystemOptions options = + GcsFileSystemOptions.builder() + .setGcsClientOptions(TEST_GCS_CLIENT_OPTIONS) + .setHnsApiEnabled(true) + .build(); + when(mockClient.isHnsBucket(TEST_BUCKET)).thenThrow(new IOException("test exception")); + + try (GcsFileSystemImpl gcsFileSystem = new GcsFileSystemImpl(mockClient, options)) { + UncheckedIOException exception = + assertThrows( + UncheckedIOException.class, () -> gcsFileSystem.resolveStrategy(TEST_BUCKET)); + + assertThat(exception).hasCauseThat().isInstanceOf(IOException.class); + assertThat(exception).hasCauseThat().hasMessageThat().isEqualTo("test exception"); + } + } + @SuppressWarnings("unchecked") private List getRegisteredTelemetryListeners(Telemetry telemetry) { try { diff --git a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemOptionsTest.java b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemOptionsTest.java index 0b4a3ff0..1514f07e 100644 --- a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemOptionsTest.java +++ b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/GcsFileSystemOptionsTest.java @@ -32,13 +32,15 @@ void createFromOptions_withValidProperties_shouldCreateCorrectOptions() { ImmutableMap.of( "fs.gs.project-id", "test-project", "fs.gs.client.type", "GRPC_CLIENT", - "fs.gs.analytics-core.read.thread.count", "32"); + "fs.gs.analytics-core.read.thread.count", "32", + "fs.gs.analytics-core.hierarchical.namespace.enable", "true"); GcsFileSystemOptions options = GcsFileSystemOptions.createFromOptions(properties, "fs.gs."); assertThat(options.getGcsClientOptions().getProjectId().get()).isEqualTo("test-project"); assertThat(options.getClientType()).isEqualTo(GcsFileSystemOptions.ClientType.GRPC_CLIENT); assertThat(options.getReadThreadCount()).isEqualTo(32); + assertThat(options.isHnsApiEnabled()).isTrue(); } @Test @@ -62,4 +64,32 @@ void createFromOptions_cacheProperties_createsCorrectOptions() { assertThat(cacheOptions.isSmallObjectCacheEnabled()).isTrue(); assertThat(cacheOptions.getSmallObjectCacheMaxSizeBytes()).isEqualTo(200 * MB); } + + @Test + void createFromOptions_withDefaultProperties_shouldCreateCorrectOptions() { + ImmutableMap properties = ImmutableMap.of(); + + GcsFileSystemOptions options = GcsFileSystemOptions.createFromOptions(properties, "fs.gs."); + + assertThat(options.getGcsClientOptions().getProjectId().isEmpty()).isTrue(); + assertThat(options.getClientType()).isEqualTo(GcsFileSystemOptions.ClientType.HTTP_CLIENT); + assertThat(options.getReadThreadCount()).isEqualTo(16); + assertThat(options.isHnsApiEnabled()).isTrue(); + + GcsCacheOptions cacheOptions = options.getGcsCacheOptions(); + assertThat(cacheOptions.isFooterCacheEnabled()).isFalse(); + assertThat(cacheOptions.getFooterCacheMaxSizeBytes()).isEqualTo(100 * MB); + assertThat(cacheOptions.isSmallObjectCacheEnabled()).isFalse(); + assertThat(cacheOptions.getSmallObjectCacheMaxSizeBytes()).isEqualTo(200 * MB); + } + + @Test + void createFromOptions_withListParallelEnabledFalse_createsCorrectOptions() { + ImmutableMap properties = + ImmutableMap.of("fs.gs.analytics-core.list.parallel.enabled", "false"); + + GcsFileSystemOptions options = GcsFileSystemOptions.createFromOptions(properties, "fs.gs."); + + assertThat(options.isListParallelEnabled()).isFalse(); + } } diff --git a/client/src/test/java/com/google/cloud/gcs/analyticscore/client/LazyExecutorServiceTest.java b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/LazyExecutorServiceTest.java new file mode 100644 index 00000000..99a76ef5 --- /dev/null +++ b/client/src/test/java/com/google/cloud/gcs/analyticscore/client/LazyExecutorServiceTest.java @@ -0,0 +1,263 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed 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. + */ + +package com.google.cloud.gcs.analyticscore.client; + +import static com.google.common.truth.Truth.assertThat; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.util.Collections; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CancellationException; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class LazyExecutorServiceTest { + + private LazyExecutorService executorService; + private AtomicBoolean executed; + + @BeforeEach + void setUp() { + executorService = new LazyExecutorService(); + executed = new AtomicBoolean(false); + } + + private String createCallableTask() { + executed.set(true); + return "success"; + } + + private void createRunnableTask() { + executed.set(true); + } + + /** + * Tests that submitting a Callable is lazy (does not execute on submit), executes on the caller's + * thread when get() is invoked, and multiple get() calls only execute the task once (verifying + * the !isDone() check). + */ + @Test + void submitCallable_isLazyAndExecutesOnceOnCallerThread() throws Exception { + AtomicReference executionThread = new AtomicReference<>(); + AtomicInteger executionCount = new AtomicInteger(0); + Callable task = + () -> { + executionCount.incrementAndGet(); + executionThread.set(Thread.currentThread()); + return "success"; + }; + Future future = executorService.submit(task); + boolean executedBeforeGet = executionCount.get() > 0; + + String result1 = future.get(10, SECONDS); + String result2 = future.get(); + + assertThat(executedBeforeGet).isFalse(); + assertThat(result1).isEqualTo("success"); + assertThat(result2).isEqualTo("success"); + assertThat(executionCount.get()).isEqualTo(1); + assertThat(executionThread.get()).isEqualTo(Thread.currentThread()); + } + + /** + * Tests that submitting a Runnable is lazy, and multiple get() calls execute the task exactly + * once. + */ + @Test + void submitRunnable_isLazyAndExecutesOnce() throws Exception { + Future future = executorService.submit(this::createRunnableTask); + boolean executedBeforeGet = executed.get(); + + future.get(); + future.get(10, SECONDS); + + assertThat(executedBeforeGet).isFalse(); + assertThat(executed.get()).isTrue(); + } + + /** + * Tests that if the executor is shut down before a task's get() is called, the task is implicitly + * cancelled and get() throws CancellationException. + */ + @Test + void shutdown_throwsCancellationExceptionOnGet() { + Future future = executorService.submit(this::createCallableTask); + + executorService.shutdown(); + + assertThat(executorService.isShutdown()).isTrue(); + assertThrows(CancellationException.class, () -> future.get(10, SECONDS)); + assertThrows(CancellationException.class, future::get); + assertThat(executed.get()).isFalse(); + assertThat(future.isCancelled()).isTrue(); + assertThat(future.isDone()).isTrue(); + } + + /** + * Tests that shutdownNow cancels pending futures similarly to shutdown, and returns an empty list + * since tasks are not queued internally. + */ + @Test + void shutdownNow_cancelsTasksAndReturnsEmptyList() { + Future future = executorService.submit(this::createCallableTask); + + List unexecutedTasks = executorService.shutdownNow(); + + assertThat(unexecutedTasks).isEmpty(); + assertThat(executorService.isShutdown()).isTrue(); + assertThrows(CancellationException.class, future::get); + assertThrows(CancellationException.class, () -> future.get(10, SECONDS)); + assertThat(executed.get()).isFalse(); + assertThat(future.isCancelled()).isTrue(); + assertThat(future.isDone()).isTrue(); + } + + @Test + void awaitTermination_andIsTerminated() throws Exception { + assertThat(executorService.isTerminated()).isFalse(); + assertThat(executorService.awaitTermination(10, SECONDS)).isFalse(); + + executorService.shutdown(); + + boolean terminated = executorService.awaitTermination(10, SECONDS); + + assertThat(executorService.isTerminated()).isTrue(); + assertThat(terminated).isTrue(); + } + + @Test + void execute_throwsRejectedExecutionException() { + assertThrows(RejectedExecutionException.class, () -> executorService.execute(() -> {})); + } + + /** + * Tests that if a task completes before shutdown, subsequent get() calls still return the + * successful result instead of throwing CancellationException. + */ + @Test + void completedTask_returnsResultAfterShutdown() throws Exception { + Future future = executorService.submit(this::createCallableTask); + future.get(); + + executorService.shutdown(); + + assertThat(future.get()).isEqualTo("success"); + assertThat(future.get(10, SECONDS)).isEqualTo("success"); + } + + @Test + void submitRunnable_withResult() throws Exception { + Future future = executorService.submit(this::createRunnableTask, "success"); + boolean executedBeforeGet = executed.get(); + + String result1 = future.get(); + String result2 = future.get(10, SECONDS); + + assertThat(executedBeforeGet).isFalse(); + assertThat(result1).isEqualTo("success"); + assertThat(result2).isEqualTo("success"); + assertThat(executed.get()).isTrue(); + } + + @Test + void submitNullTask_throwsNullPointerException() { + assertThrows(NullPointerException.class, () -> executorService.submit((Callable) null)); + assertThrows(NullPointerException.class, () -> executorService.submit((Runnable) null)); + assertThrows(NullPointerException.class, () -> executorService.submit((Runnable) null, "res")); + } + + @Test + void submitAfterShutdown_throwsRejectedExecutionException() { + executorService.shutdown(); + + assertThrows(RejectedExecutionException.class, () -> executorService.submit(() -> "task")); + assertThrows(RejectedExecutionException.class, () -> executorService.submit(() -> {})); + } + + /** + * Tests that explicitly cancelling a future prevents it from executing, which is natively handled + * by FutureTask's state checks when run() is invoked. + */ + @Test + void cancel_preventsTaskExecution() { + Future future = executorService.submit(this::createCallableTask); + + future.cancel(true); + + assertThrows(CancellationException.class, future::get); + assertThrows(CancellationException.class, () -> future.get(10, SECONDS)); + assertThat(executed.get()).isFalse(); + } + + @Test + void get_whenThreadInterrupted_throwsInterruptedException() { + Future future = executorService.submit(this::createCallableTask); + Thread.currentThread().interrupt(); + + assertThrows(InterruptedException.class, future::get); + + // Thread.interrupted() clears the interrupt status. + assertThat(Thread.interrupted()).isFalse(); + assertThat(executed.get()).isFalse(); + } + + @Test + void getWithTimeout_whenThreadInterrupted_throwsInterruptedException() { + Future future = executorService.submit(this::createCallableTask); + Thread.currentThread().interrupt(); + + assertThrows(InterruptedException.class, () -> future.get(10, SECONDS)); + + // Thread.interrupted() clears the interrupt status. + assertThat(Thread.interrupted()).isFalse(); + assertThat(executed.get()).isFalse(); + } + + @Test + void getWithTimeout_whenTimeoutZeroOrNegative_throwsTimeoutException() { + Future future = executorService.submit(this::createCallableTask); + + assertThrows(TimeoutException.class, () -> future.get(0, SECONDS)); + assertThrows(TimeoutException.class, () -> future.get(-1, SECONDS)); + + assertThat(executed.get()).isFalse(); + } + + @Test + void invokeMethods_throwUnsupportedOperationException() { + assertThrows( + UnsupportedOperationException.class, + () -> executorService.invokeAll(Collections.emptyList())); + assertThrows( + UnsupportedOperationException.class, + () -> executorService.invokeAll(Collections.emptyList(), 10, SECONDS)); + assertThrows( + UnsupportedOperationException.class, + () -> executorService.invokeAny(Collections.emptyList())); + assertThrows( + UnsupportedOperationException.class, + () -> executorService.invokeAny(Collections.emptyList(), 10, SECONDS)); + } +} diff --git a/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/LoggingOpenTelemetryProviderTest.java b/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/LoggingOpenTelemetryProviderTest.java index f51f5c57..6ceba672 100644 --- a/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/LoggingOpenTelemetryProviderTest.java +++ b/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/LoggingOpenTelemetryProviderTest.java @@ -23,7 +23,7 @@ class LoggingOpenTelemetryProviderTest { @Test - void get_returnsNonNullInstance() { + void testGet_returnsNonNullInstance() { try (LoggingOpenTelemetryProvider provider = new LoggingOpenTelemetryProvider(OpenTelemetryOptions.builder().build())) { OpenTelemetry openTelemetry = provider.getOpenTelemetry(); @@ -33,7 +33,7 @@ void get_returnsNonNullInstance() { } @Test - void get_returnsSameInstanceOnMultipleCalls() { + void testGet_returnsSameInstanceOnMultipleCalls() { try (LoggingOpenTelemetryProvider provider = new LoggingOpenTelemetryProvider(OpenTelemetryOptions.builder().build())) { OpenTelemetry firstCall = provider.getOpenTelemetry(); @@ -44,7 +44,7 @@ void get_returnsSameInstanceOnMultipleCalls() { } @Test - void constructor_withDuration_createsSuccessfully() { + void testConstructor_withDuration_createsSuccessfully() { try (LoggingOpenTelemetryProvider provider = new LoggingOpenTelemetryProvider( OpenTelemetryOptions.builder().setExportIntervalSeconds(30).build())) { diff --git a/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/LoggingTelemetryReporterTest.java b/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/LoggingTelemetryReporterTest.java index 9d474c27..85aadc83 100644 --- a/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/LoggingTelemetryReporterTest.java +++ b/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/LoggingTelemetryReporterTest.java @@ -26,7 +26,7 @@ class LoggingTelemetryReporterTest { @Test - void formatMetrics_singleMetricWithoutAttributes() { + public void testFormatMetrics_singleMetricWithoutAttributes() { try (LoggingTelemetryReporter reporter = new LoggingTelemetryReporter(LoggingTelemetryOptions.builder().build())) { Map metrics = @@ -43,7 +43,7 @@ void formatMetrics_singleMetricWithoutAttributes() { } @Test - void formatMetrics_singleMetricWithAttributes() { + public void testFormatMetrics_singleMetricWithAttributes() { try (LoggingTelemetryReporter reporter = new LoggingTelemetryReporter(LoggingTelemetryOptions.builder().build())) { Map metrics = @@ -64,7 +64,7 @@ void formatMetrics_singleMetricWithAttributes() { } @Test - void formatMetrics_multipleMetrics() { + public void testFormatMetrics_multipleMetrics() { try (LoggingTelemetryReporter reporter = new LoggingTelemetryReporter(LoggingTelemetryOptions.builder().build())) { Map metrics = diff --git a/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/OpenTelemetryReporterTest.java b/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/OpenTelemetryReporterTest.java index b0cdcd23..e652c41d 100644 --- a/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/OpenTelemetryReporterTest.java +++ b/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/OpenTelemetryReporterTest.java @@ -67,7 +67,7 @@ void setUp() { } @Test - void operationEnd_recordsMetrics() { + void testOperationEnd_recordsMetrics() { OpenTelemetryOptions options = OpenTelemetryOptions.builder() .setEnabled(true) diff --git a/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/TelemetryOptionsTest.java b/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/TelemetryOptionsTest.java index 0b2b8b2a..60c4d32e 100644 --- a/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/TelemetryOptionsTest.java +++ b/common/src/test/java/com/google/cloud/gcs/analyticscore/common/telemetry/TelemetryOptionsTest.java @@ -22,10 +22,10 @@ import java.util.Map; import org.junit.jupiter.api.Test; -class TelemetryOptionsTest { +public class TelemetryOptionsTest { @Test - void builderWithCustomTelemetryOptions() { + public void testBuilderWithCustomTelemetryOptions() { OperationListener listener = new OperationListener() { @Override @@ -44,7 +44,7 @@ public void onOperationEnd(Operation operation, java.util.Map m } @Test - void createFromOptions_Empty() { + public void testCreateFromOptions_Empty() { Map optionsMap = new HashMap<>(); TelemetryOptions options = TelemetryOptions.createFromOptions(optionsMap, "prefix."); @@ -53,7 +53,7 @@ void createFromOptions_Empty() { } @Test - void createFromOptions_WithLogging() { + public void testCreateFromOptions_WithLogging() { Map optionsMap = new HashMap<>(); optionsMap.put("prefix.telemetry.logging.enabled", "true"); optionsMap.put("prefix.telemetry.logging.level", "INFO"); @@ -68,7 +68,7 @@ void createFromOptions_WithLogging() { } @Test - void createFromOptions_WithOpenTelemetry() { + public void testCreateFromOptions_WithOpenTelemetry() { Map optionsMap = new HashMap<>(); optionsMap.put("prefix.telemetry.opentelemetry.enabled", "true"); optionsMap.put("prefix.telemetry.opentelemetry.provider-type", "LOGGING"); @@ -83,7 +83,7 @@ void createFromOptions_WithOpenTelemetry() { } @Test - void createFromOptions_WithAll() { + public void testCreateFromOptions_WithAll() { Map optionsMap = new HashMap<>(); optionsMap.put("prefix.telemetry.logging.enabled", "true"); optionsMap.put("prefix.telemetry.opentelemetry.enabled", "false"); diff --git a/test-lib/src/main/java/com/google/cloud/gcs/analyticscore/client/FakeGcsClientImpl.java b/test-lib/src/main/java/com/google/cloud/gcs/analyticscore/client/FakeGcsClientImpl.java index 81c17521..69efabf2 100644 --- a/test-lib/src/main/java/com/google/cloud/gcs/analyticscore/client/FakeGcsClientImpl.java +++ b/test-lib/src/main/java/com/google/cloud/gcs/analyticscore/client/FakeGcsClientImpl.java @@ -51,6 +51,12 @@ protected Storage createStorage(Optional credentials) { return storage; } + @Override + BucketProperties getBucketProperties(String bucketName) { + // FakeStorageRpc does not support bucket operations + return BucketProperties.create(false); + } + @Override public VectoredSeekableByteChannel openReadChannel( GcsItemInfo itemInfo, GcsReadOptions readOptions) throws IOException {