sunchao commented on code in PR #7363:
URL: https://github.com/apache/datafusion-comet/pull/7363#discussion_r4485814103
##########
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:
Could this wait for the asynchronous close to finish before returning, and
likewise drain pending closes before shared fixture counters are reset?
Releasing the latch lets the dispatcher thread proceed but does not wait for
its `closeCount` increment. That increment can therefore arrive after the next
test resets the shared provider state.
I repeatedly ran the unmodified `aBlockingCloseDoesNotHoldUpRelease` and
`releasingTheLastReferenceClosesTheProvider` through JUnit. At iteration 321,
the blocking-close test failed at its `closeCount == 0` assertion with actual 1
from an earlier close. Draining the dispatcher executor between the same tests
passed all 4000 executions. A completion latch or executor barrier in cleanup
would keep the non-blocking-release assertion while removing this test-order
race.
##########
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:
Could this architecture description match the new scan-only behavior?
`CometIcebergNativeWrite` no longer calls `withVendedCredentialProvider`, and
the user guide now correctly says a refresh endpoint makes native writes fall
back to Spark. This paragraph still says the provider is selected for the
native scan and writer. The following "Each adapter" paragraph also implies
Spark 3.x and 4.x implementations for all three adapters, although this
vended-credentials adapter exists only on Spark 4.x. Please qualify both
statements consistently.
--
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]
