Add option to enable jdk support for UDS

Author: sarahchen6

src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java

@@ -96,6 +96,7 @@ String tag() {
     public static final int SOCKET_BUFFER_BYTES = -1;
     public static final boolean DEFAULT_BLOCKING = false;
     public static final boolean DEFAULT_ENABLE_TELEMETRY = true;
+    public static final boolean DEFAULT_ENABLE_JDK_SOCKET = true;
 
     public static final boolean DEFAULT_ENABLE_AGGREGATION = true;
     public static final boolean DEFAULT_ENABLE_ORIGIN_DETECTION = true;
@@ -243,12 +244,7 @@ public NonBlockingStatsDClient(final NonBlockingStatsDClientBuilder builder)
         externalEnv = originDetectionEnabled ? Utf8.sanitize(System.getenv("DD_EXTERNAL_ENV")) : "";
 
         try {
-            clientChannel =
-                    createByteChannel(
-                            builder.addressLookup,
-                            builder.timeout,
-                            builder.connectionTimeout,
-                            builder.socketBufferSize);
+            clientChannel = createByteChannel(addressLookup, timeout, connectionTimeout, bufferSize, enableJdkSocket);
 
             ThreadFactory threadFactory =
                     builder.threadFactory != null
@@ -291,12 +287,7 @@ public NonBlockingStatsDClient(final NonBlockingStatsDClientBuilder builder)
                 telemetryClientChannel = clientChannel;
                 telemetryStatsDProcessor = statsDProcessor;
             } else {
-                telemetryClientChannel =
-                        createByteChannel(
-                                builder.telemetryAddressLookup,
-                                builder.timeout,
-                                builder.connectionTimeout,
-                                builder.socketBufferSize);
+                telemetryClientChannel = createByteChannel(telemetryAddressLookup, timeout, connectionTimeout, bufferSize, enableJdkSocket);
 
                 // similar settings, but a single worker and non-blocking.
                 telemetryStatsDProcessor =
@@ -479,10 +470,7 @@ StringBuilder tagString(final String[] tags, StringBuilder builder) {
     }
 
     ClientChannel createByteChannel(
-            Callable<SocketAddress> addressLookup,
-            int timeout,
-            int connectionTimeout,
-            int bufferSize)
+            Callable<SocketAddress> addressLookup, int timeout, int connectionTimeout, int bufferSize, boolean enableJdkSocket)
             throws Exception {
         final SocketAddress address = addressLookup.call();
         if (address instanceof NamedPipeSocketAddress) {
@@ -496,8 +484,7 @@ ClientChannel createByteChannel(
             // Allow us to support `unix://` for both kind of sockets like in go.
             switch (unixAddr.getTransportType()) {
                 case UDS_STREAM:
-                    return new UnixStreamClientChannel(
-                            unixAddr.getAddress(), timeout, connectionTimeout, bufferSize);
+                    return new UnixStreamClientChannel(unixAddr.getAddress(), timeout, connectionTimeout, bufferSize, enableJdkSocket);
                 case UDS_DATAGRAM:
                 case UDS:
                     return new UnixDatagramClientChannel(

src/main/java/com/timgroup/statsd/NonBlockingStatsDClientBuilder.java

@@ -57,6 +57,9 @@ public class NonBlockingStatsDClientBuilder implements Cloneable {
 
     public boolean enableAggregation = NonBlockingStatsDClient.DEFAULT_ENABLE_AGGREGATION;
 
+    /** Enable native JDK support for UDS. */
+    public boolean enableJdkSocket = NonBlockingStatsDClient.DEFAULT_ENABLE_JDK_SOCKET;
+
     /** Telemetry flush interval, in milliseconds. */
     public int telemetryFlushInterval = Telemetry.DEFAULT_FLUSH_INTERVAL;
 
@@ -327,6 +330,11 @@ public NonBlockingStatsDClientBuilder originDetectionEnabled(boolean val) {
         return this;
     }
 
+    public NonBlockingStatsDClientBuilder enableJdkSocket(boolean val) {
+        enableJdkSocket = val;
+        return this;
+    }
+
     /**
      * Request that all metrics from this client to be enriched to specified tag cardinality.
      *
@@ -529,7 +537,7 @@ protected static Callable<SocketAddress> staticUnixResolution(
             @Override public SocketAddress call() {
                 SocketAddress socketAddress;
                 // Use native UDS support for compatible Java versions and jnr-unixsocket support otherwise.
-                if (VersionUtils.isJavaVersionAtLeast(16)) {
+                if (VersionUtils.isJavaVersionAtLeast(16) && NonBlockingStatsDClient.DEFAULT_ENABLE_JDK_SOCKET) {
                     try {
                         // Use reflection to avoid compiling Java 16+ classes in incompatible versions
                         Class<?> unixDomainSocketAddressClass = Class.forName("java.net.UnixDomainSocketAddress");

src/main/java/com/timgroup/statsd/UnixStreamClientChannel.java

@@ -17,6 +17,7 @@ public class UnixStreamClientChannel implements ClientChannel {
     private final int timeout;
     private final int connectionTimeout;
     private final int bufferSize;
+    private final boolean enableJdkSocket;
 
     private SocketChannel delegate;
     private final ByteBuffer delimiterBuffer =
@@ -27,16 +28,15 @@ public class UnixStreamClientChannel implements ClientChannel {
      *
      * @param address Location of named pipe
      */
-    UnixStreamClientChannel(
-            SocketAddress address, int timeout, int connectionTimeout, int bufferSize)
-            throws IOException {
+    UnixStreamClientChannel(SocketAddress address, int timeout, int connectionTimeout, int bufferSize, boolean enableJdkSocket) throws IOException {
         this.delegate = null;
         this.address = address;
         System.out.println("========== Constructor address: " + address);
         System.out.println("========== Constructor address type: " + address.getClass().getName());
         this.timeout = timeout;
         this.connectionTimeout = connectionTimeout;
         this.bufferSize = bufferSize;
+        this.enableJdkSocket = enableJdkSocket;
     }
 
     @Override
@@ -133,7 +133,7 @@ private void connect() throws IOException {
 
         long deadline = System.nanoTime() + connectionTimeout * 1_000_000L;
         // Use native UDS support for compatible Java versions and jnr-unixsocket support otherwise.
-        if (VersionUtils.isJavaVersionAtLeast(16)) {
+        if (VersionUtils.isJavaVersionAtLeast(16) && enableJdkSocket) {
             try {
                 // Use reflection to avoid compiling Java 16+ classes in incompatible versions
                 Class<?> protocolFamilyClass = Class.forName("java.net.StandardProtocolFamily");
@@ -180,7 +180,7 @@ private void connect() throws IOException {
                 throw new IOException("Failed to create UnixStreamClientChannel for native UDS implementation", e);
             }
         }
-        // Default to jnr-unixsocket if Java version is less than 16
+        // Default to jnr-unixsocket if Java version is less than 16 or native UDS support is disabled
         UnixSocketChannel channel = UnixSocketChannel.create();
 
         if (connectionTimeout > 0) {

src/test/java/com/timgroup/statsd/NonBlockingStatsDClientTest.java

@@ -1285,7 +1285,8 @@ ClientChannel createByteChannel(
                                     Callable<SocketAddress> addressLookup,
                                     int timeout,
                                     int connectionTimeout,
-                                    int bufferSize)
+                                    int bufferSize,
+                                    boolean enableJdkSocket)
                                     throws Exception {
                                 return new DatagramClientChannel(addressLookup.call()) {
                                     @Override
@@ -1336,7 +1337,8 @@ ClientChannel createByteChannel(
                                     Callable<SocketAddress> addressLookup,
                                     int timeout,
                                     int connectionTimeout,
-                                    int bufferSize)
+                                    int bufferSize,
+                                    boolean enableJdkSocket)
                                     throws Exception {
                                 return new DatagramClientChannel(addressLookup.call()) {
                                     @Override