Added capability to use (and enforce) native netty transport in azure-cosmos-spark

sdk/cosmos/azure-cosmos-spark_3-1_2-12/CHANGELOG.md

@@ -3,6 +3,7 @@
 ### 4.30.0-beta.1 (Unreleased)
 
 #### Features Added
+* Added capability to use (and enforce) native netty transport. The native transport is more efficient - esepcially when the number of TCP connections being used is high. - See [PR 39834](https://github.com/Azure/azure-sdk-for-java/pull/39834)
 
 #### Breaking Changes
 

sdk/cosmos/azure-cosmos-spark_3-2_2-12/CHANGELOG.md

@@ -3,6 +3,7 @@
 ### 4.30.0-beta.1 (Unreleased)
 
 #### Features Added
+* Added capability to use (and enforce) native netty transport. The native transport is more efficient - esepcially when the number of TCP connections being used is high. - See [PR 39834](https://github.com/Azure/azure-sdk-for-java/pull/39834)
 
 #### Breaking Changes
 

sdk/cosmos/azure-cosmos-spark_3-3_2-12/CHANGELOG.md

@@ -3,6 +3,7 @@
 ### 4.30.0-beta.1 (Unreleased)
 
 #### Features Added
+* Added capability to use (and enforce) native netty transport. The native transport is more efficient - esepcially when the number of TCP connections being used is high. - See [PR 39834](https://github.com/Azure/azure-sdk-for-java/pull/39834)
 
 #### Breaking Changes
 

sdk/cosmos/azure-cosmos-spark_3-4_2-12/CHANGELOG.md

@@ -3,6 +3,7 @@
 ### 4.30.0-beta.1 (Unreleased)
 
 #### Features Added
+* Added capability to use (and enforce) native netty transport. The native transport is more efficient - esepcially when the number of TCP connections being used is high. - See [PR 39834](https://github.com/Azure/azure-sdk-for-java/pull/39834)
 
 #### Breaking Changes
 

sdk/cosmos/azure-cosmos-spark_3-5_2-12/CHANGELOG.md

@@ -3,6 +3,7 @@
 ### 4.30.0-beta.1 (Unreleased)
 
 #### Features Added
+* Added capability to use (and enforce) native netty transport. The native transport is more efficient - esepcially when the number of TCP connections being used is high. - See [PR 39834](https://github.com/Azure/azure-sdk-for-java/pull/39834)
 
 #### Breaking Changes
 

sdk/cosmos/azure-cosmos-spark_3_2-12/docs/configuration-reference.md

@@ -24,6 +24,7 @@
 | Config Property Name                                              | Default | Description                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
 |:------------------------------------------------------------------|:--------|:--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| 
 | `spark.cosmos.useGatewayMode`                                     | `false` | Use gateway mode for the client operations                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    |
+| `spark.cosmos.enforceNativeTransport`                             | `false` | Indicates whether native netty transport should be enforced. If true, on client creation an exception will be thrown if netty is not able to use native transport. The native transport is more efficient especially when there is a high number of connections.                                                                                                                                                                                                                                                                                                                                                |
 | `spark.cosmos.http.connectionPoolSize`                            | `1000`  | Gateway mode connection pool size                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             |
 | `spark.cosmos.read.forceEventualConsistency`                      | `true`  | Makes the client use Eventual consistency for read operations instead of using the default account level consistency                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
 | `spark.cosmos.applicationName`                                    | None    | Application name                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              |

sdk/cosmos/azure-cosmos-spark_3_2-12/pom.xml

@@ -24,6 +24,7 @@
     <maven.compiler.target>11</maven.compiler.target>
     <jacoco.skip>true</jacoco.skip>
     <shadingPrefix>azure_cosmos_spark</shadingPrefix>
+    <shadingPrefixNetty>azurecosmosspark</shadingPrefixNetty>
     <legal>
       <![CDATA[[INFO] Any downloads listed may be third party software.  Microsoft grants you no rights for third party software.]]>
     </legal>
@@ -226,6 +227,7 @@
                 <include>org.scala-lang.modules:scala-java8-compat_2.12:[0.8.0]</include> <!-- {x-include-update;cosmos_org.scala-lang.modules:scala-java8-compat_2.12;external_dependency} -->
                 <include>io.projectreactor:reactor-scala-extensions_2.12:[0.8.0]</include> <!-- {x-include-update;cosmos_io.projectreactor:reactor-scala-extensions_2.12;external_dependency} -->
                 <include>org.scalatest:scalatest_2.12:[3.2.2]</include> <!-- {x-include-update;cosmos_org.scalatest:scalatest_2.12;external_dependency} -->
+                <include>org.apache.maven.plugins:maven-antrun-plugin:[3.1.0]</include> <!-- {x-include-update;org.apache.maven.plugins:maven-antrun-plugin;external_dependency} -->
                 <include>net.alchim31.maven:scala-maven-plugin:[4.8.1]</include> <!-- {x-include-update;cosmos_net.alchim31.maven:scala-maven-plugin;external_dependency} -->
                 <include>org.scalastyle:scalastyle-maven-plugin:[1.0.0]</include> <!-- {x-include-update;cosmos_org.scalastyle:scalastyle-maven-plugin;external_dependency} -->
                 <include>com.fasterxml.jackson.datatype:jackson-datatype-jsr310:[2.13.5]</include> <!-- {x-include-update;com.fasterxml.jackson.datatype:jackson-datatype-jsr310;external_dependency} -->
@@ -301,7 +303,7 @@
             <version>3.3.0</version> <!-- {x-version-update;org.apache.maven.plugins:maven-jar-plugin;external_dependency} -->
             <executions>
               <execution>
-                <id>empty-javadoc-jar-with-readme</id>
+                <id>01-empty-javadoc-jar-with-readme</id>
                 <phase>package</phase>
                 <goals>
                   <goal>jar</goal>
@@ -313,33 +315,6 @@
               </execution>
             </executions>
           </plugin>
-          <plugin>
-            <groupId>org.apache.maven.plugins</groupId>
-            <artifactId>maven-antrun-plugin</artifactId>
-            <version>3.1.0</version> <!-- {x-version-update;org.apache.maven.plugins:maven-antrun-plugin;external_dependency} -->
-            <executions>
-              <execution>
-                <id>copy-readme-to-javadocTemp</id>
-                <phase>prepare-package</phase>
-                <configuration>
-                  <target>
-                    <echo>Deleting existing ${project.basedir}/javadocTemp</echo>
-                    <delete includeEmptyDirs="true" quiet="true">
-                      <fileset dir="${project.basedir}/javadocTemp"/>
-                    </delete>
-                    <echo>Copying ${project.basedir}/README.md to
-                      ${project.basedir}/javadocTemp/README.md
-                    </echo>
-                    <copy file="${project.basedir}/README.md" tofile="${project.basedir}/javadocTemp/README.md"/>
-                  </target>
-                </configuration>
-                <goals>
-                  <goal>run</goal>
-                </goals>
-              </execution>
-            </executions>
-          </plugin>
-          <!-- END: Empty Java Doc -->
           <plugin>
             <groupId>net.alchim31.maven</groupId>
             <artifactId>scala-maven-plugin</artifactId>
@@ -399,7 +374,7 @@
             <version>3.5.2</version> <!-- {x-version-update;org.apache.maven.plugins:maven-shade-plugin;external_dependency} -->
             <executions>
               <execution>
-                <id>shade</id>
+                <id>02-shade</id>
                 <phase>package</phase>
                 <goals>
                   <goal>shade</goal>
@@ -427,7 +402,7 @@
                     </relocation>
                     <relocation>
                       <pattern>io.netty</pattern>
-                      <shadedPattern>${shadingPrefix}.io.netty</shadedPattern>
+                      <shadedPattern>${shadingPrefixNetty}.io.netty</shadedPattern>
                     </relocation>
                     <relocation>
                       <pattern>org.codehaus</pattern>
@@ -604,6 +579,54 @@
               </execution>
             </executions>
           </plugin>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-antrun-plugin</artifactId>
+            <version>3.1.0</version> <!-- {x-version-update;org.apache.maven.plugins:maven-antrun-plugin;external_dependency} -->
+            <executions>
+              <execution>
+                <id>01-copy-readme-to-javadocTemp</id>
+                <phase>prepare-package</phase>
+                <configuration>
+                  <target>
+                    <echo>Deleting existing ${project.basedir}/javadocTemp</echo>
+                    <delete includeEmptyDirs="true" quiet="true">
+                      <fileset dir="${project.basedir}/javadocTemp"/>
+                    </delete>
+                    <echo>Copying ${project.basedir}/README.md to
+                      ${project.basedir}/javadocTemp/README.md
+                    </echo>
+                    <copy file="${project.basedir}/README.md" tofile="${project.basedir}/javadocTemp/README.md"/>
+                  </target>
+                </configuration>
+                <goals>
+                  <goal>run</goal>
+                </goals>
+              </execution>
+              <execution>
+                <id>03-repack</id>
+                <phase>package</phase>
+                <goals>
+                  <goal>run</goal>
+                </goals>
+                <configuration>
+                  <target>
+                    <unzip src="${project.build.directory}/${project.build.finalName}.jar" dest="${project.build.directory}/tmp" />
+                    <move file="${project.build.directory}/tmp/META-INF/native/libnetty_resolver_dns_native_macos_x86_64.jnilib" tofile="${project.build.directory}/tmp/META-INF/native/libazurecosmosspark_netty_resolver_dns_native_macos_x86_64.jnilib" />
+                    <move file="${project.build.directory}/tmp/META-INF/native/libnetty_tcnative_linux_aarch_64.so" tofile="${project.build.directory}/tmp/META-INF/native/libazurecosmosspark_netty_tcnative_linux_aarch_64.so" />
+                    <move file="${project.build.directory}/tmp/META-INF/native/libnetty_tcnative_linux_x86_64.so" tofile="${project.build.directory}/tmp/META-INF/native/libazurecosmosspark_netty_tcnative_linux_x86_64.so" />
+                    <move file="${project.build.directory}/tmp/META-INF/native/libnetty_tcnative_osx_aarch_64.jnilib" tofile="${project.build.directory}/tmp/META-INF/native/libazurecosmosspark_netty_tcnative_osx_aarch_64.jnilib" />
+                    <move file="${project.build.directory}/tmp/META-INF/native/libnetty_tcnative_osx_x86_64.jnilib" tofile="${project.build.directory}/tmp/META-INF/native/libazurecosmosspark_netty_tcnative_osx_x86_64.jnilib" />
+                    <move file="${project.build.directory}/tmp/META-INF/native/libnetty_transport_native_epoll_x86_64.so" tofile="${project.build.directory}/tmp/META-INF/native/libazurecosmosspark_netty_transport_native_epoll_x86_64.so" />
+                    <move file="${project.build.directory}/tmp/META-INF/native/libnetty_transport_native_kqueue_x86_64.jnilib" tofile="${project.build.directory}/tmp/META-INF/native/libazurecosmosspark_netty_transport_native_kqueue_x86_64.jnilib" />
+                    <move file="${project.build.directory}/tmp/META-INF/native/netty_tcnative_windows_x86_64.dll" tofile="${project.build.directory}/tmp/META-INF/native/azurecosmosspark_netty_tcnative_windows_x86_64.dll" />
+                    <zip basedir="${project.build.directory}/tmp" destfile="${project.build.directory}/${project.build.finalName}.jar" />
+                    <delete dir="${project.build.directory}/tmp" />
+                  </target>
+                </configuration>
+              </execution>
+            </executions>
+          </plugin>
           <plugin>
             <groupId>org.codehaus.mojo</groupId>
             <artifactId>xml-maven-plugin</artifactId>

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/spark/CosmosClientCache.scala

@@ -174,6 +174,15 @@ private[spark] object CosmosClientCache extends BasicLoggingTrait {
   // scalastyle:off cyclomatic.complexity
   private[this] def createCosmosAsyncClient(cosmosClientConfiguration: CosmosClientConfiguration,
                                             cosmosClientStateHandle: Option[CosmosClientMetadataCachesSnapshot]): CosmosAsyncClient = {
+      if (cosmosClientConfiguration.enforceNativeTransport && !io.netty.channel.epoll.Epoll.isAvailable) {
+        throw new IllegalStateException(
+          "The enforcement of native transport is enabled in your configuration and native transport is not " +
+            "available. Either ensure `spark.cosmos.enforceNativeTransport` is set to false or make " +
+            "sure you use a Spark environment supporting native transport.",
+          io.netty.channel.epoll.Epoll.unavailabilityCause()
+        )
+      }
+
       var builder = new CosmosClientBuilder()
           .endpoint(cosmosClientConfiguration.endpoint)
           .userAgentSuffix(cosmosClientConfiguration.applicationName)

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/spark/CosmosClientConfiguration.scala

@@ -14,6 +14,7 @@ private[spark] case class CosmosClientConfiguration (
                                                       customApplicationNameSuffix: Option[String],
                                                       applicationName: String,
                                                       useGatewayMode: Boolean,
+                                                      enforceNativeTransport: Boolean,
                                                       proactiveConnectionInitialization: Option[String],
                                                       proactiveConnectionInitializationDurationInSeconds: Int,
                                                       httpConnectionPoolSize: Int,
@@ -69,6 +70,7 @@ private[spark] object CosmosClientConfiguration {
       customApplicationNameSuffix,
       applicationName,
       cosmosAccountConfig.useGatewayMode,
+      cosmosAccountConfig.enforceNativeTransport,
       cosmosAccountConfig.proactiveConnectionInitialization,
       cosmosAccountConfig.proactiveConnectionInitializationDurationInSeconds,
       cosmosAccountConfig.httpConnectionPoolSize,

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/spark/CosmosConfig.scala

@@ -59,6 +59,7 @@ private[spark] object CosmosConfigNames {
   val DisableTcpConnectionEndpointRediscovery = "spark.cosmos.disableTcpConnectionEndpointRediscovery"
   val ApplicationName = "spark.cosmos.applicationName"
   val UseGatewayMode = "spark.cosmos.useGatewayMode"
+  val EnforceNativeTransport = "spark.cosmos.enforceNativeTransport"
   val ProactiveConnectionInitialization = "spark.cosmos.proactiveConnectionInitialization"
   val ProactiveConnectionInitializationDurationInSeconds = "spark.cosmos.proactiveConnectionInitializationDurationInSeconds"
   val GatewayConnectionPoolSize = "spark.cosmos.http.connectionPoolSize"
@@ -151,6 +152,7 @@ private[spark] object CosmosConfigNames {
     DisableTcpConnectionEndpointRediscovery,
     ApplicationName,
     UseGatewayMode,
+    EnforceNativeTransport,
     ProactiveConnectionInitialization,
     ProactiveConnectionInitializationDurationInSeconds,
     GatewayConnectionPoolSize,
@@ -328,6 +330,7 @@ private case class CosmosAccountConfig(endpoint: String,
                                        accountName: String,
                                        applicationName: Option[String],
                                        useGatewayMode: Boolean,
+                                       enforceNativeTransport: Boolean,
                                        proactiveConnectionInitialization: Option[String],
                                        proactiveConnectionInitializationDurationInSeconds: Int,
                                        httpConnectionPoolSize: Int,
@@ -408,6 +411,12 @@ private object CosmosAccountConfig {
     parseFromStringFunction = useGatewayMode => useGatewayMode.toBoolean,
     helpMessage = "Use gateway mode for the client operations")
 
+  private val EnforceNativeTransport = CosmosConfigEntry[Boolean](key = CosmosConfigNames.EnforceNativeTransport,
+    mandatory = false,
+    defaultValue = Some(false),
+    parseFromStringFunction = enforceNativeTransport => enforceNativeTransport.toBoolean,
+    helpMessage = "Flag indicating whether native Netty transport availability should be enforced.")
+
   private val ProactiveConnectionInitialization = CosmosConfigEntry[String](key = CosmosConfigNames.ProactiveConnectionInitialization,
     mandatory = false,
     defaultValue = None,
@@ -502,6 +511,7 @@ private object CosmosAccountConfig {
     val accountName = CosmosConfigEntry.parse(cfg, CosmosAccountName)
     val applicationName = CosmosConfigEntry.parse(cfg, ApplicationName)
     val useGatewayMode = CosmosConfigEntry.parse(cfg, UseGatewayMode)
+    val enforceNativeTransport = CosmosConfigEntry.parse(cfg, EnforceNativeTransport)
     val proactiveConnectionInitialization = CosmosConfigEntry.parse(cfg, ProactiveConnectionInitialization)
     val proactiveConnectionInitializationDurationInSeconds = CosmosConfigEntry.parse(cfg, ProactiveConnectionInitializationDurationInSeconds)
     val httpConnectionPoolSize = CosmosConfigEntry.parse(cfg, HttpConnectionPoolSize)
@@ -558,6 +568,7 @@ private object CosmosAccountConfig {
       accountName.get,
       applicationName,
       useGatewayMode.get,
+      enforceNativeTransport.get,
       proactiveConnectionInitialization,
       proactiveConnectionInitializationDurationInSeconds.get,
       httpConnectionPoolSize.get,