laxmanmamididd commented on code in PR #7363:
URL: https://github.com/apache/datafusion-comet/pull/7363#discussion_r4503365849
##########
spark/src/test/java/org/apache/comet/cloud/s3/CometS3CredentialDispatcherTest.java:
##########
@@ -267,4 +282,145 @@ public void closeAllSwallowsProviderExceptions() {
// Both close() invocations ran even though one (the first to be visited)
threw.
assertEquals(2, TestCometS3CredentialProvider.closeCount.get());
}
+
+ @Test
+ public void releasingTheLastReferenceClosesTheProvider() throws Exception {
+ long h1 = init();
+ long h2 = init();
+ assertEquals(h1, h2);
+
+ CometS3CredentialDispatcher.release(h1);
+ assertEquals(0, TestCometS3CredentialProvider.closeCount.get());
+ CometS3CredentialDispatcher.getCredentialsForPath(h2, "b", "k", READ);
+
+ CometS3CredentialDispatcher.release(h2);
+ assertEquals(1, TestCometS3CredentialProvider.closeCount.get());
+ assertEquals(0, CometS3CredentialDispatcher.registeredCount());
+ Exception thrown =
+ assertThrows(
+ IllegalStateException.class,
+ () -> CometS3CredentialDispatcher.getCredentialsForPath(h2, "b",
"k", READ));
+ assertTrue(thrown.getMessage().contains("not initialized"));
+ }
+
+ @Test
+ public void aRetiredKeyGetsANewProvider() {
+ long retired = init();
+ CometS3CredentialDispatcher.release(retired);
+
+ long fresh = init();
+
+ assertNotEquals(retired, fresh);
+ assertEquals(2, TestCometS3CredentialProvider.initCount.get());
+ assertEquals(1, CometS3CredentialDispatcher.registeredCount());
+ }
+
+ /** An extra release, or one for a handle closed at shutdown, changes
nothing. */
+ @Test
+ public void unknownOrRepeatedReleasesAreIgnored() {
+ CometS3CredentialDispatcher.release(Long.MAX_VALUE);
+ long handle = init();
+ CometS3CredentialDispatcher.release(handle);
+ CometS3CredentialDispatcher.release(handle);
+
+ assertEquals(1, TestCometS3CredentialProvider.closeCount.get());
+ long other = init();
+ CometS3CredentialDispatcher.release(handle);
+ assertEquals(1, CometS3CredentialDispatcher.registeredCount());
+ CometS3CredentialDispatcher.release(other);
+ assertEquals(2, TestCometS3CredentialProvider.closeCount.get());
+ }
+
+ @Test
+ public void concurrentAcquireAndReleaseNeverUseAClosedProvider() throws
Exception {
+ int threads = 8;
+ int rounds = 2000;
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ try {
+ List<Future<?>> futures = new ArrayList<>();
+ for (int t = 0; t < threads; t++) {
+ futures.add(
+ pool.submit(
+ () -> {
+ for (int i = 0; i < rounds; i++) {
+ long handle = init();
+ CometS3CredentialDispatcher.getCredentialsForPath(handle,
"b", "k", READ);
+ CometS3CredentialDispatcher.release(handle);
+ }
+ return null;
+ }));
+ }
+ for (Future<?> f : futures) {
+ f.get();
+ }
+ } finally {
+ pool.shutdownNow();
+ }
+ assertEquals(0, CometS3CredentialDispatcher.registeredCount());
+ assertEquals(
+ TestCometS3CredentialProvider.initCount.get(),
+ TestCometS3CredentialProvider.closeCount.get());
+ }
+
+ /** The releasing thread may be a native reader's, so the close runs on the
dispatcher's own. */
+ @Test
+ public void releaseClosesOnTheDispatcherThread() throws Exception {
+ CometS3CredentialDispatcher.setCloserForTesting(null);
+ TestCometS3CredentialProvider.closeEntered = new CountDownLatch(1);
+ long handle = init();
+
+ CometS3CredentialDispatcher.release(handle);
+
+ assertTrue(TestCometS3CredentialProvider.closeEntered.await(10,
TimeUnit.SECONDS));
+ assertEquals("comet-s3-credential-close",
TestCometS3CredentialProvider.lastCloseThread);
+ assertEquals(0, CometS3CredentialDispatcher.registeredCount());
+ }
+
+ /** A close that blocks holds up neither the releasing thread nor a new
reader for the key. */
+ @Test
+ public void aBlockingCloseDoesNotHoldUpRelease() throws Exception {
+ CometS3CredentialDispatcher.setCloserForTesting(null);
+ CountDownLatch block = new CountDownLatch(1);
+ TestCometS3CredentialProvider.blockClose = block;
+ TestCometS3CredentialProvider.closeEntered = new CountDownLatch(1);
+ try {
+ long retired = init();
+ CometS3CredentialDispatcher.release(retired);
+ assertTrue(TestCometS3CredentialProvider.closeEntered.await(10,
TimeUnit.SECONDS));
+
+ long fresh = init();
+ CometS3CredentialDispatcher.getCredentialsForPath(fresh, "b", "k", READ);
+ assertNotEquals(retired, fresh);
+ assertEquals(0, TestCometS3CredentialProvider.closeCount.get());
+ } finally {
+ block.countDown();
Review Comment:
Thanks for the repro. Fixed in fc3a4f2a1 on #7464, now under this branch
too. The dispatcher has a test-only `awaitQueuedClosesForTesting()`, which
queues a no-op on its single close thread and waits for it, so every close
queued before the call has finished. `@Before` calls it before `closeAll()` and
the counter reset, and `@After` calls it after restoring the default closer. So
no close can land in the next test's counts, whatever ordering the tests run
in. `aBlockingCloseDoesNotHoldUpRelease` still checks that release doesn't wait
for a blocked close, since the drain only runs in cleanup.
Running the whole class 1,500 times in a row through JUnit gave no failed
runs.
##########
docs/source/contributor-guide/s3-credential-provider-design.md:
##########
@@ -158,20 +158,29 @@ The Parquet path forwards the full `fs.s3a.*` config
subset as `catalog_properti
## Built-in adapters
-Comet ships two reference SPI implementations under
`org.apache.comet.cloud.s3`, so standard provider classes that the native Rust
list does not match work with a config change instead of bespoke code:
+Comet ships three SPI implementations under `org.apache.comet.cloud.s3`, so
standard provider classes that the native Rust list does not match work with a
config change instead of bespoke code:
- `HadoopS3ACredentialProviderAdapter` delegates to Hadoop S3A's own provider
construction (`S3AUtils.createAWSCredentialProviderSet` on Hadoop 3.3.x,
`CredentialProviderListFactory.createAWSCredentialProviderList` on 3.4+).
- `AwsSdkCredentialProviderAdapter` wraps a raw AWS SDK provider named in
`fs.s3a.comet.credential.adapter.class`.
+- `IcebergVendedS3CredentialProvider` resolves REST-vended Iceberg credentials
through Iceberg's `AwsClientProperties`, as `S3FileIO` does, and refreshes them
through `client.refresh-credentials-endpoint`. It exists only in the spark-4.x
source set. `CometIcebergNativeScan.withVendedCredentialProvider` selects it
for the native Iceberg scan and writer when the catalog returns a refresh
endpoint and no provider is configured.
Review Comment:
Updated in 9122cca8a. The paragraph already said scan-only after 286d711c8.
The "Each adapter" paragraph now says that only the Hadoop and AWS SDK adapters
have spark-3.x and spark-4.x bodies, and that
`IcebergVendedS3CredentialProvider` has only the spark-4.x body, so Spark 3.x
builds neither ship nor select it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
