resolve comments

xinlian12 · Mar 29, 2024 · 8f819f9 · 8f819f9
1 parent 4b486fd
commit 8f819f9
Show file tree

Hide file tree

Showing 6 changed files with 12 additions and 17 deletions.
diff --git a/eng/code-quality-reports/src/main/resources/checkstyle/checkstyle-suppressions.xml b/eng/code-quality-reports/src/main/resources/checkstyle/checkstyle-suppressions.xml
@@ -318,8 +318,8 @@ the main ServiceBusClientBuilder. -->
   <suppress checks="ConstantName" files="com.azure.spring.cloud.config.AppConfigurationPropertySourceLocator"/>
   <suppress checks="com.azure.tools.checkstyle.checks.GoodLoggingCheck"
             files="[/\\]azure-cosmos-kafka-connect[/\\]"/>
-  <suppress checks="com.azure.tools.checkstyle.checks.ExternalDependencyExposedCheck" files="com.azure.cosmos.kafka.connect.CosmosDBSourceConnector"/>
-  <suppress checks="com.azure.tools.checkstyle.checks.ExternalDependencyExposedCheck" files="com.azure.cosmos.kafka.connect.CosmosDBSinkConnector"/>
+  <suppress checks="com.azure.tools.checkstyle.checks.ExternalDependencyExposedCheck" files="com.azure.cosmos.kafka.connect.CosmosSourceConnector"/>
+  <suppress checks="com.azure.tools.checkstyle.checks.ExternalDependencyExposedCheck" files="com.azure.cosmos.kafka.connect.CosmosSinkConnector"/>
 
   <!-- Checkstyle suppressions for resource manager package -->
   <suppress checks="com.azure.tools.checkstyle.checks.ServiceClientCheck" files="com.azure.resourcemanager.*"/>

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) {