fixKafkaConnectorStuckIssue (#46378)

* fix connector stuck issue

---------

Co-authored-by: annie-mac <[email protected]>

Author: xinlian12

sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md

@@ -7,6 +7,7 @@
 #### Breaking Changes
 
 #### Bugs Fixed
+* Fixed an issue where `CosmosSourceConnector` got stuck when restart - See [PR 46378](https://github.com/Azure/azure-sdk-for-java/pull/46378)
 
 #### Other Changes
 

sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/CosmosSinkConnector.java

@@ -5,6 +5,7 @@
 
 import com.azure.cosmos.CosmosAsyncContainer;
 import com.azure.cosmos.CosmosAsyncDatabase;
+import com.azure.cosmos.implementation.ImplementationBridgeHelpers;
 import com.azure.cosmos.implementation.UUIDs;
 import com.azure.cosmos.implementation.apachecommons.lang.RandomUtils;
 import com.azure.cosmos.implementation.apachecommons.lang.StringUtils;
@@ -51,6 +52,11 @@ public final class CosmosSinkConnector extends SinkConnector implements AutoClos
     private String connectorName;
     private CosmosClientCacheItem cosmosClientItem;
 
+    static {
+        //initialize all accessors from different threads can cause deadlock issues, so here we force loading ahead of time
+        ImplementationBridgeHelpers.initializeAllAccessors();
+    }
+
     @Override
     public void start(Map<String, String> props) {
         LOGGER.info("Starting the kafka cosmos sink connector");

sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/CosmosSourceConnector.java

@@ -80,6 +80,11 @@ public final class CosmosSourceConnector extends SourceConnector implements Auto
     private IMetadataReader metadataReader;
     private String connectorName;
 
+    static {
+        //initialize all accessors from different threads can cause deadlock issues, so here we force loading ahead of time
+        ImplementationBridgeHelpers.initializeAllAccessors();
+    }
+
     @Override
     public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> connectorConfig) {
         return ExactlyOnceSupport.SUPPORTED;

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ImplementationBridgeHelpers.java

@@ -102,7 +102,7 @@
 public class ImplementationBridgeHelpers {
     private final static Logger logger = LoggerFactory.getLogger(ImplementationBridgeHelpers.class);
 
-    private static void  initializeAllAccessors() {
+    public static void initializeAllAccessors() {
         ModelBridgeInternal.initializeAllAccessors();
         UtilBridgeInternal.initializeAllAccessors();
         BridgeInternal.initializeAllAccessors();