core: drop non-blocking DNS resolver from S3ClientCache

S3ClientCache opted the AWS SDK Netty async client into a non-blocking
DNS resolver, gated on a hardcoded probe for the unrelocated class
io.netty.resolver.dns.DnsAddressResolverGroup. On the Spark 4 / Core ETL
driver bundle this probe is misleading: Spark and Hadoop put the
unrelocated class on the classpath, so the probe returns true, but the
active AWS SDK comes from iceberg-aws-bundle, whose NettyNioAsyncHttpClient
-> AwaitCloseChannelPoolMap -> BootstrapProvider calls DnsResolverLoader,
which is relocated to org.apache.iceberg.aws.shaded.io.netty.resolver.dns
-- a class the bundle omits. Enabling non-blocking DNS then fails when the
first connection pool/bootstrap is created.

Remove the setting entirely and always use the SDK default (blocking)
resolver. The benefit is negligible for this workload: the S3 backup
reader goes through a single (or few) bucket host(s) per job with
long-lived cached clients, keep-alive, and connection pooling, so DNS is
resolved at connection creation and then cached by the JVM/Netty --
steady-state lookups are rare. Dropping it removes the relocation/
missing-class fragility for no measurable loss and shrinks the surface
area to test across the OSS, aggregate, and prod-parity packaging shapes.

This deletes the class-name constant, the availability field and probe
helpers, the conditional builder branch, and the dns=nonBlocking cache-key
suffix, and removes the now-defunct test plus its classpath helper. Netty
relocation in the driver bundles remains correct and is retained.

Revisit only if a future workload reads across many distinct S3 hosts or
runs where DNS is slow.

Author: liucao-dd

cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/S3ClientCache.java

@@ -68,8 +68,6 @@ public final class S3ClientCache
     static final int CONNECTION_TIMEOUT_SECONDS = 5;
     static final int READ_TIMEOUT_SECONDS = 120;
     static final int CONNECTION_MAX_IDLE_TIME_SECONDS = 60;
-    static final String NETTY_DNS_RESOLVER_CLASS = "io.netty.resolver.dns.DnsAddressResolverGroup";
-    private static final boolean NON_BLOCKING_DNS_RESOLVER_AVAILABLE = isClassAvailable(NETTY_DNS_RESOLVER_CLASS);
     private static volatile boolean loggedTaskSlotSource = false;
 
     // Separate caches for different client types
@@ -221,7 +219,7 @@ private static String getAsyncCacheKey(S3ClientConfig config, ResolvedAsyncHttpC
                + "|connectTimeout=" + CONNECTION_TIMEOUT_SECONDS
                + "|readTimeout=" + READ_TIMEOUT_SECONDS
                + "|idleTime=" + CONNECTION_MAX_IDLE_TIME_SECONDS
-               + "|tcpKeepAlive=true|dns=nonBlocking";
+               + "|tcpKeepAlive=true";
     }
 
     // ========================================================================
@@ -295,16 +293,6 @@ private static S3AsyncClient buildS3AsyncClient(S3ClientConfig config, ResolvedA
             .connectionMaxIdleTime(Duration.ofSeconds(CONNECTION_MAX_IDLE_TIME_SECONDS))
             .tcpKeepAlive(true);
 
-        if (isNonBlockingDnsResolverAvailable())
-        {
-            httpClientBuilder.useNonBlockingDnsResolver(true);
-        }
-        else
-        {
-            LOGGER.info("Netty non-blocking DNS resolver is not available on the classpath; "
-                        + "building S3AsyncClient with the AWS SDK default DNS resolver");
-        }
-
         software.amazon.awssdk.services.s3.S3AsyncClientBuilder builder = S3AsyncClient.builder()
             .region(Region.of(config.s3Region()))
             .credentialsProvider(credentialsProvider)
@@ -317,35 +305,16 @@ private static S3AsyncClient buildS3AsyncClient(S3ClientConfig config, ResolvedA
         }
 
         LOGGER.info("Built S3AsyncClient region={} maxConcurrency={} maxPendingConnectionAcquires={} "
-                    + "nonBlockingDnsResolver={} (knob={}, taskSlotsSource={}, taskSlots={})",
+                    + "(knob={}, taskSlotsSource={}, taskSlots={})",
                     config.s3Region(),
                     resolved.maxConcurrency,
                     resolved.maxPendingConnectionAcquires,
-                    isNonBlockingDnsResolverAvailable(),
                     config.s3HttpMaxConcurrency(),
                     resolved.taskSlots.source,
                     resolved.taskSlots.taskSlots);
         return builder.build();
     }
 
-    static boolean isNonBlockingDnsResolverAvailable()
-    {
-        return NON_BLOCKING_DNS_RESOLVER_AVAILABLE;
-    }
-
-    private static boolean isClassAvailable(String className)
-    {
-        try
-        {
-            Class.forName(className, false, S3ClientCache.class.getClassLoader());
-            return true;
-        }
-        catch (ClassNotFoundException exception)
-        {
-            return false;
-        }
-    }
-
     static int resolveMaxConcurrency(S3ClientConfig config)
     {
         int maxConcurrency = config.s3HttpMaxConcurrency();

cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/data/S3ClientCacheTest.java

@@ -223,13 +223,6 @@ void testResolveMaxPendingConnectionAcquires()
         assertThat(S3ClientCache.resolveMaxPendingConnectionAcquires(2000)).isEqualTo(2048);
     }
 
-    @Test
-    void testNonBlockingDnsResolverAvailabilityMatchesClasspath()
-    {
-        assertThat(S3ClientCache.isNonBlockingDnsResolverAvailable())
-        .isEqualTo(isClassAvailable(S3ClientCache.NETTY_DNS_RESOLVER_CLASS));
-    }
-
     @Test
     void testResolveTaskSlotsPrioritizesSparkConfThenEnvThenRuntime()
     {
@@ -395,17 +388,4 @@ private S3ClientConfig createConfig(Map<String, String> options)
     {
         return S3ClientConfig.create(new CaseInsensitiveStringMap(options));
     }
-
-    private static boolean isClassAvailable(String className)
-    {
-        try
-        {
-            Class.forName(className, false, S3ClientCacheTest.class.getClassLoader());
-            return true;
-        }
-        catch (ClassNotFoundException exception)
-        {
-            return false;
-        }
-    }
 }