Fixing test regressions (Azure#45798)

Author: FabianMeiswinkel

sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSourceConnectorTest.java

@@ -756,7 +756,7 @@ private void setupDefaultConnectorInternalStatesWithMetadataKafkaReader(
             CosmosClientCache.getCosmosClient(
                 cosmosSourceConfig.getAccountConfig(),
                 "testKafkaConnector");
-        KafkaCosmosReflectionUtils.setCosmosClient(sourceConnector, clientCacheItem.getClient());
+        KafkaCosmosReflectionUtils.setCosmosClientCacheItem(sourceConnector, clientCacheItem);
 
         InMemoryStorageReader inMemoryStorageReader = new InMemoryStorageReader();
         MetadataKafkaStorageManager metadataReader = new MetadataKafkaStorageManager(inMemoryStorageReader);
@@ -792,7 +792,7 @@ private void setupDefaultConnectorInternalStatesWithMetadataCosmosReader(
             CosmosClientCache.getCosmosClient(
                 cosmosSourceConfig.getAccountConfig(),
                 "testKafkaConnector");
-        KafkaCosmosReflectionUtils.setCosmosClient(sourceConnector, clientCacheItem.getClient());
+        KafkaCosmosReflectionUtils.setCosmosClientCacheItem(sourceConnector, clientCacheItem);
 
         CosmosAsyncContainer container = clientCacheItem.getClient().getDatabase(databaseName).getContainer(containerName);
         MetadataCosmosStorageManager cosmosStorageManager = new MetadataCosmosStorageManager(container);

sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/KafkaCosmosReflectionUtils.java

@@ -4,6 +4,7 @@
 package com.azure.cosmos.kafka.connect;
 
 import com.azure.cosmos.CosmosAsyncClient;
+import com.azure.cosmos.kafka.connect.implementation.CosmosClientCacheItem;
 import com.azure.cosmos.kafka.connect.implementation.sink.CosmosSinkTask;
 import com.azure.cosmos.kafka.connect.implementation.source.CosmosSourceConfig;
 import com.azure.cosmos.kafka.connect.implementation.source.IMetadataReader;
@@ -31,8 +32,8 @@ private static <T> T get(Object object, String fieldName) {
         }
     }
 
-    public static void setCosmosClient(CosmosSourceConnector sourceConnector, CosmosAsyncClient cosmosAsyncClient) {
-        set(sourceConnector, cosmosAsyncClient,"cosmosClient");
+    public static void setCosmosClientCacheItem(CosmosSourceConnector sourceConnector, CosmosClientCacheItem clientCacheItem) {
+        set(sourceConnector, clientCacheItem,"cosmosClientItem");
     }
 
     public static void setCosmosSourceConfig(CosmosSourceConnector sourceConnector, CosmosSourceConfig sourceConfig) {
@@ -58,7 +59,8 @@ public static void setMetadataMonitorThread(
     }
 
     public static CosmosAsyncClient getCosmosClient(CosmosSourceConnector sourceConnector) {
-        return get(sourceConnector,"cosmosClient");
+        CosmosClientCacheItem clientCacheItem = get(sourceConnector,"cosmosClientItem");
+        return clientCacheItem.getClient();
     }
 
     public static MetadataKafkaStorageManager getKafkaOffsetStorageReader(CosmosSourceConnector sourceConnector) {
@@ -74,7 +76,8 @@ public static void setSinkTaskContext(CosmosSinkTask sinkTask, SinkTaskContext s
     }
 
     public static CosmosAsyncClient getSinkTaskCosmosClient(CosmosSinkTask sinkTask) {
-        return get(sinkTask,"cosmosClient");
+        CosmosClientCacheItem clientCacheItem = get(sinkTask,"cosmosClientItem");
+        return clientCacheItem.getClient();
     }
 
     public static void setConnectorName(CosmosSourceConnector sourceConnector, String connectorName) {