resolve comments

xinlian12 · Mar 29, 2024 · 6decd72 · 6decd72
1 parent 4b486fd
commit 6decd72
Show file tree

Hide file tree

Showing 5 changed files with 10 additions and 15 deletions.
diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/pom.xml b/sdk/cosmos/azure-cosmos-kafka-connect/pom.xml
@@ -44,16 +44,16 @@ Licensed under the MIT License.
     <revapi.skip>true</revapi.skip>
     <!-- Configures the Java 9+ run to perform the required module exports, opens, and reads that are necessary for testing but shouldn't be part of the module-info. -->
     <javaModulesSurefireArgLine>
+      --add-opens com.azure.cosmos/com.azure.cosmos.implementation=ALL-UNNAMED
+      --add-opens com.azure.cosmos/com.azure.cosmos.implementation.apachecommons.lang=ALL-UNNAMED
+      --add-opens com.azure.cosmos/com.azure.cosmos.implementation.caches=ALL-UNNAMED
+      --add-opens com.azure.cosmos/com.azure.cosmos.implementation.faultinjection=ALL-UNNAMED
+      --add-opens com.azure.cosmos/com.azure.cosmos.implementation.routing=ALL-UNNAMED
       --add-opens com.azure.cosmos.kafka.connect/com.azure.cosmos.kafka.connect=ALL-UNNAMED
       --add-opens com.azure.cosmos.kafka.connect/com.azure.cosmos.kafka.connect.implementation=ALL-UNNAMED
-      --add-opens com.azure.cosmos.kafka.connect/com.azure.cosmos.kafka.connect.implementation.source=com.fasterxml.jackson.databind,ALL-UNNAMED
       --add-opens com.azure.cosmos.kafka.connect/com.azure.cosmos.kafka.connect.implementation.sink=ALL-UNNAMED
       --add-opens com.azure.cosmos.kafka.connect/com.azure.cosmos.kafka.connect.implementation.sink.idStrategy=ALL-UNNAMED
-      --add-opens com.azure.cosmos/com.azure.cosmos.implementation=ALL-UNNAMED
-      --add-opens com.azure.cosmos/com.azure.cosmos.implementation.routing=ALL-UNNAMED
-      --add-opens com.azure.cosmos/com.azure.cosmos.implementation.caches=ALL-UNNAMED
-      --add-opens com.azure.cosmos/com.azure.cosmos.implementation.faultinjection=ALL-UNNAMED
-      --add-opens com.azure.cosmos/com.azure.cosmos.implementation.apachecommons.lang=ALL-UNNAMED
+      --add-opens com.azure.cosmos.kafka.connect/com.azure.cosmos.kafka.connect.implementation.source=com.fasterxml.jackson.databind,ALL-UNNAMED
 
     </javaModulesSurefireArgLine>
   </properties>

diff --git a/...osmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/CosmosSinkConnector.java b/...osmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/CosmosSinkConnector.java
@@ -48,7 +48,7 @@ public List<Map<String, String>> taskConfigs(int maxTasks) {
 
     @Override
     public void stop() {
-        LOGGER.debug("Kafka Cosmos sink connector {} is stopped.");
+        LOGGER.info("Kafka Cosmos sink connector {} is stopped.");
     }
 
     @Override

diff --git a/...ct/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosSchedulers.java b/...ct/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosSchedulers.java
@@ -7,7 +7,7 @@
 import reactor.core.scheduler.Schedulers;
 
 public class KafkaCosmosSchedulers {
-    private static final String SINK_BOUNDED_ELASTIC_THREAD_NAME = "sink-bounded-elastic";
+    private static final String SINK_BOUNDED_ELASTIC_THREAD_NAME = "kafka-cosmos-sink-bounded-elastic";
     private static final int TTL_FOR_SCHEDULER_WORKER_IN_SECONDS = 60; // same as BoundedElasticScheduler.DEFAULT_TTL_SECONDS
     public static final Scheduler SINK_BOUNDED_ELASTIC = Schedulers.newBoundedElastic(
         Schedulers.DEFAULT_BOUNDED_ELASTIC_SIZE,

diff --git a/...nect/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosSinkTask.java b/...nect/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosSinkTask.java
@@ -32,7 +32,7 @@ public String version() {
 
     @Override
     public void start(Map<String, String> props) {
-        LOGGER.info("Starting the kafka cosmos sink task...");
+        LOGGER.info("Starting the kafka cosmos sink task");
         this.sinkTaskConfig = new CosmosSinkTaskConfig(props);
         this.cosmosClient = CosmosClientStore.getCosmosClient(this.sinkTaskConfig.getAccountConfig());
         this.sinkRecordTransformer = new SinkRecordTransformer(this.sinkTaskConfig);
@@ -85,11 +85,9 @@ record -> this.sinkTaskConfig
 
     @Override
     public void stop() {
-        LOGGER.info("Stopping Kafka CosmosDB sink task...");
+        LOGGER.info("Stopping Kafka CosmosDB sink task");
         if (this.cosmosClient != null) {
             this.cosmosClient.close();
         }
-
-        this.cosmosClient = null;
     }
 }
diff --git a/...rc/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosWriteException.java b/...rc/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosWriteException.java
@@ -9,9 +9,6 @@
  * Generic CosmosDb sink write exceptions.
  */
 public class CosmosWriteException extends ConnectException {
-    /**
-     *
-     */
     private static final long serialVersionUID = 1L;
 
     public CosmosWriteException(String message) {