diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md b/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md
index 628e9057d60e..673becf9cbfd 100644
--- a/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md
+++ b/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md
@@ -9,6 +9,7 @@
 #### Bugs Fixed
 
 #### Other Changes
+* Added more logs in `CosmosSourceTask` and `CosmosSinkTask` - See [PR 46224](https://github.com/Azure/azure-sdk-for-java/pull/46224)
 
 ### 2.4.0 (2025-06-24)
 
diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosSinkTask.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosSinkTask.java
index a8a6be01755b..4ad34a2d1d35 100644
--- a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosSinkTask.java
+++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosSinkTask.java
@@ -35,28 +35,36 @@ public String version() {
     @Override
     public void start(Map<String, String> props) {
         LOGGER.info("Starting the kafka cosmos sink task");
-        this.sinkTaskConfig = new CosmosSinkTaskConfig(props);
-        this.cosmosClientItem =
-            CosmosClientCache.getCosmosClient(
-                this.sinkTaskConfig.getAccountConfig(),
-                this.sinkTaskConfig.getTaskId(),
-                this.sinkTaskConfig.getClientMetadataCachesSnapshot());
-        LOGGER.info("The taskId is " + this.sinkTaskConfig.getTaskId());
-        this.throughputControlClientItem = this.getThroughputControlCosmosClient();
-        this.sinkRecordTransformer = new SinkRecordTransformer(this.sinkTaskConfig);
-
-        if (this.sinkTaskConfig.getWriteConfig().isBulkEnabled()) {
-            this.cosmosWriter =
-                new CosmosBulkWriter(
-                    this.sinkTaskConfig.getWriteConfig(),
-                    this.sinkTaskConfig.getThroughputControlConfig(),
-                    this.context.errantRecordReporter());
-        } else {
-            this.cosmosWriter =
-                new CosmosPointWriter(
-                    this.sinkTaskConfig.getWriteConfig(),
-                    this.sinkTaskConfig.getThroughputControlConfig(),
-                    context.errantRecordReporter());
+
+        try {
+            this.sinkTaskConfig = new CosmosSinkTaskConfig(props);
+            this.cosmosClientItem =
+                CosmosClientCache.getCosmosClient(
+                    this.sinkTaskConfig.getAccountConfig(),
+                    this.sinkTaskConfig.getTaskId(),
+                    this.sinkTaskConfig.getClientMetadataCachesSnapshot());
+            LOGGER.info("The taskId is " + this.sinkTaskConfig.getTaskId());
+            this.throughputControlClientItem = this.getThroughputControlCosmosClient();
+            this.sinkRecordTransformer = new SinkRecordTransformer(this.sinkTaskConfig);
+
+            if (this.sinkTaskConfig.getWriteConfig().isBulkEnabled()) {
+                this.cosmosWriter =
+                    new CosmosBulkWriter(
+                        this.sinkTaskConfig.getWriteConfig(),
+                        this.sinkTaskConfig.getThroughputControlConfig(),
+                        this.context.errantRecordReporter());
+            } else {
+                this.cosmosWriter =
+                    new CosmosPointWriter(
+                        this.sinkTaskConfig.getWriteConfig(),
+                        this.sinkTaskConfig.getThroughputControlConfig(),
+                        context.errantRecordReporter());
+            }
+        } catch (Throwable e) {
+            LOGGER.warn("Error occurred while starting the kafka sink task", e);
+            this.cleanup();
+
+            throw e;
         }
     }
 
@@ -80,7 +88,7 @@ public void put(Collection<SinkRecord> records) {
             return;
         }
 
-        LOGGER.debug("Sending {} records to be written", records.size());
+        LOGGER.info("Sending {} records to be written", records.size());
 
         // group by container
         Map<String, List<SinkRecord>> recordsByContainer =
@@ -115,17 +123,26 @@ record -> this.sinkTaskConfig
         }
     }
 
-    @Override
-    public void stop() {
-        LOGGER.info("Stopping Kafka CosmosDB sink task");
+
+    private void cleanup() {
+        LOGGER.info("Cleaning up CosmosSinkTask");
+
         if (this.throughputControlClientItem != null && this.throughputControlClientItem != this.cosmosClientItem) {
+            LOGGER.debug("Releasing throughput control cosmos client");
             CosmosClientCache.releaseCosmosClient(this.throughputControlClientItem.getClientConfig());
             this.throughputControlClientItem = null;
         }
 
         if (this.cosmosClientItem != null) {
+            LOGGER.debug("Releasing cosmos client");
             CosmosClientCache.releaseCosmosClient(this.cosmosClientItem.getClientConfig());
             this.cosmosClientItem = null;
         }
     }
+
+    @Override
+    public void stop() {
+        LOGGER.info("Stopping Kafka CosmosDB sink task");
+        this.cleanup();
+    }
 }
diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosSourceTask.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosSourceTask.java
index 90e0544ba443..f98c07ce5f22 100644
--- a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosSourceTask.java
+++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosSourceTask.java
@@ -52,21 +52,36 @@ public String version() {
     public void start(Map<String, String> map) {
         LOGGER.info("Starting the kafka cosmos source task...");
 
-        this.taskConfig = new CosmosSourceTaskConfig(map);
-        if (this.taskConfig.getMetadataTaskUnit() != null) {
-            // adding metadata task units into the head of the queue
-            this.taskUnitsQueue.add(this.taskConfig.getMetadataTaskUnit());
-        }
-
-        this.taskUnitsQueue.addAll(this.taskConfig.getFeedRangeTaskUnits());
-        LOGGER.info("Creating the cosmos client");
+        try {
+            LOGGER.info("Resetting task queue");
+            this.taskUnitsQueue.clear();
+
+            this.taskConfig = new CosmosSourceTaskConfig(map);
+            if (this.taskConfig.getMetadataTaskUnit() != null) {
+                // adding metadata task units into the head of the queue
+                LOGGER.info("Adding metadata task to task {}", this.taskConfig.getTaskId());
+                this.taskUnitsQueue.add(this.taskConfig.getMetadataTaskUnit());
+            }
 
-        this.cosmosClientItem =
-            CosmosClientCache.getCosmosClient(
-                this.taskConfig.getAccountConfig(),
-                this.taskConfig.getTaskId(),
-                this.taskConfig.getCosmosClientMetadataCachesSnapshot());
-        this.throughputControlCosmosClientItem = this.getThroughputControlCosmosClientItem();
+            LOGGER.info(
+                "Adding {} feed range tasks to task {}",
+                this.taskConfig.getFeedRangeTaskUnits().size(),
+                this.taskConfig.getTaskId());
+
+            this.taskUnitsQueue.addAll(this.taskConfig.getFeedRangeTaskUnits());
+            LOGGER.info("Creating the cosmos client");
+
+            this.cosmosClientItem =
+                CosmosClientCache.getCosmosClient(
+                    this.taskConfig.getAccountConfig(),
+                    this.taskConfig.getTaskId(),
+                    this.taskConfig.getCosmosClientMetadataCachesSnapshot());
+            this.throughputControlCosmosClientItem = this.getThroughputControlCosmosClientItem();
+        } catch (Throwable ex) {
+            LOGGER.warn("Failed to start the cosmos source task", ex);
+            this.cleanup();
+            throw ex;
+        }
     }
 
     private CosmosClientCacheItem getThroughputControlCosmosClientItem() {
@@ -113,7 +128,7 @@ public List<SourceRecord> poll() {
                 }
 
                 stopwatch.stop();
-                LOGGER.debug(
+                LOGGER.info(
                     "Return {} records, databaseName {}, containerName {}, containerRid {}, feedRange {}, durationInMs {}",
                     results.size(),
                     ((FeedRangeTaskUnit) taskUnit).getDatabaseName(),
@@ -127,6 +142,8 @@ public List<SourceRecord> poll() {
         } catch (Exception e) {
             // for error cases, we should always put the task back to the queue
             this.taskUnitsQueue.add(taskUnit);
+            LOGGER.warn("Polling task failed", e);
+
             throw KafkaCosmosExceptionsHelper.convertToConnectException(e, "PollTask failed");
         }
     }
@@ -390,16 +407,25 @@ private CosmosChangeFeedRequestOptions getChangeFeedRequestOptions(FeedRangeTask
         return changeFeedRequestOptions;
     }
 
-    @Override
-    public void stop() {
+    private void cleanup() {
+        LOGGER.info("Cleaning up CosmosSourceTask");
+
         if (this.throughputControlCosmosClientItem != null && this.throughputControlCosmosClientItem != this.cosmosClientItem) {
+            LOGGER.debug("Releasing throughput control cosmos client");
             CosmosClientCache.releaseCosmosClient(this.throughputControlCosmosClientItem.getClientConfig());
             this.throughputControlCosmosClientItem = null;
         }
 
         if (this.cosmosClientItem != null) {
+            LOGGER.debug("Releasing cosmos client");
             CosmosClientCache.releaseCosmosClient(this.cosmosClientItem.getClientConfig());
             this.cosmosClientItem = null;
         }
     }
+
+    @Override
+    public void stop() {
+        LOGGER.info("Stopping CosmosSourceTask");
+        this.cleanup();
+    }
 }