From c5f190c363586fc0ba84e3585ba6099c095dfada Mon Sep 17 00:00:00 2001 From: annie-mac Date: Wed, 30 Jul 2025 13:38:16 -0700 Subject: [PATCH 1/6] add more logs in kafka connector --- .../implementation/sink/CosmosSinkTask.java | 71 ++++++++++++------- .../source/CosmosSourceTask.java | 53 +++++++++----- 2 files changed, 80 insertions(+), 44 deletions(-) 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..83f01f1f0b5f 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 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 (Exception e) { + LOGGER.warn("Error occurred while starting the kafka sink task", e); + this.cleanup(); + + throw e; } } @@ -75,12 +83,12 @@ private CosmosClientCacheItem getThroughputControlCosmosClient() { @Override public void put(Collection records) { - if (records == null || records.isEmpty()) { + if (records == null) { LOGGER.debug("No records to be written"); return; } - LOGGER.debug("Sending {} records to be written", records.size()); + LOGGER.info("Sending {} records to be written", records.size()); // group by container Map> 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..907491ed4f7f 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 @@ -51,22 +51,30 @@ public String version() { @Override public void start(Map map) { LOGGER.info("Starting the kafka cosmos source task..."); + 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 - this.taskUnitsQueue.add(this.taskConfig.getMetadataTaskUnit()); - } - - this.taskUnitsQueue.addAll(this.taskConfig.getFeedRangeTaskUnits()); - LOGGER.info("Creating the cosmos client"); + try { + 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.cosmosClientItem = - CosmosClientCache.getCosmosClient( - this.taskConfig.getAccountConfig(), - this.taskConfig.getTaskId(), - this.taskConfig.getCosmosClientMetadataCachesSnapshot()); - this.throughputControlCosmosClientItem = this.getThroughputControlCosmosClientItem(); + 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 (Exception ex) { + LOGGER.warn("Failed to start the cosmos source task", ex); + this.cleanup(); + throw ex; + } } private CosmosClientCacheItem getThroughputControlCosmosClientItem() { @@ -113,7 +121,7 @@ public List poll() { } stopwatch.stop(); - LOGGER.debug( + LOGGER.info( "Return {} records, databaseName {}, containerName {}, containerRid {}, feedRange {}, durationInMs {}", results.size(), ((FeedRangeTaskUnit) taskUnit).getDatabaseName(), @@ -127,6 +135,8 @@ public List 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 +400,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(); + } } From 2676f8347229706898219439b9bdff8b75e2718e Mon Sep 17 00:00:00 2001 From: annie-mac Date: Wed, 30 Jul 2025 13:40:38 -0700 Subject: [PATCH 2/6] add extra logs for kafka connector --- .../connect/implementation/source/CosmosSourceTask.java | 6 ++++++ 1 file changed, 6 insertions(+) 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 907491ed4f7f..621c6b43785c 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 @@ -58,9 +58,15 @@ public void start(Map map) { 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()); } + 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"); From 85f245207e47ac5d7e1078712be628a8226d6a0e Mon Sep 17 00:00:00 2001 From: annie-mac Date: Wed, 30 Jul 2025 14:03:56 -0700 Subject: [PATCH 3/6] resolve comments --- .../connect/implementation/sink/CosmosSinkTask.java | 4 ++-- .../connect/implementation/source/CosmosSourceTask.java | 9 +++++---- 2 files changed, 7 insertions(+), 6 deletions(-) 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 83f01f1f0b5f..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 @@ -60,7 +60,7 @@ public void start(Map props) { this.sinkTaskConfig.getThroughputControlConfig(), context.errantRecordReporter()); } - } catch (Exception e) { + } catch (Throwable e) { LOGGER.warn("Error occurred while starting the kafka sink task", e); this.cleanup(); @@ -83,7 +83,7 @@ private CosmosClientCacheItem getThroughputControlCosmosClient() { @Override public void put(Collection records) { - if (records == null) { + if (records == null || records.isEmpty()) { LOGGER.debug("No records to be written"); return; } 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 621c6b43785c..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 @@ -51,10 +51,11 @@ public String version() { @Override public void start(Map map) { LOGGER.info("Starting the kafka cosmos source task..."); - LOGGER.info("Resetting task queue"); - this.taskUnitsQueue.clear(); 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 @@ -66,7 +67,7 @@ public void start(Map map) { "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"); @@ -76,7 +77,7 @@ public void start(Map map) { this.taskConfig.getTaskId(), this.taskConfig.getCosmosClientMetadataCachesSnapshot()); this.throughputControlCosmosClientItem = this.getThroughputControlCosmosClientItem(); - } catch (Exception ex) { + } catch (Throwable ex) { LOGGER.warn("Failed to start the cosmos source task", ex); this.cleanup(); throw ex; From 10f416c53092a2d0f1adea3d937fc8a89d10ad6c Mon Sep 17 00:00:00 2001 From: annie-mac Date: Wed, 30 Jul 2025 14:11:41 -0700 Subject: [PATCH 4/6] try to fix readme --- sdk/cosmos/azure-cosmos/README.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/sdk/cosmos/azure-cosmos/README.md b/sdk/cosmos/azure-cosmos/README.md index 7b7b1322395c..b1efd12823d0 100644 --- a/sdk/cosmos/azure-cosmos/README.md +++ b/sdk/cosmos/azure-cosmos/README.md @@ -122,10 +122,10 @@ You may learn more about partitioning [here](https://learn.microsoft.com/azure/c ## Examples The following section provides several code snippets covering some of the most common Cosmos DB SQL API tasks, including: -* [Create Cosmos Client](#create-cosmos-client "Create Cosmos Client") -* [Create Database](#create-database "Create Database") -* [Create Container](#create-container "Create Container") -* [CRUD operation on Items](#crud-operation-on-items "CRUD operation on Items") +* [Create Cosmos Client](#create-cosmos-client) +* [Create Database](#create-database) +* [Create Container](#create-container) +* [CRUD operation on Items](#crud-operation-on-items) ### Create Cosmos Client ```java readme-sample-createCosmosClient2 From 89db87aca16e773dce67d5a2374bf6f400da457f Mon Sep 17 00:00:00 2001 From: annie-mac Date: Wed, 30 Jul 2025 14:14:48 -0700 Subject: [PATCH 5/6] update changelog --- sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) 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) From 1a2af9d00856a6b178a8ea7e8d9f123ba1434cbe Mon Sep 17 00:00:00 2001 From: annie-mac Date: Wed, 30 Jul 2025 14:38:35 -0700 Subject: [PATCH 6/6] fix more readme --- sdk/cosmos/azure-cosmos-encryption/README.md | 8 ++++---- sdk/cosmos/azure-cosmos-test/README.md | 8 ++++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/sdk/cosmos/azure-cosmos-encryption/README.md b/sdk/cosmos/azure-cosmos-encryption/README.md index 95d4b4c351e2..ab0baa204665 100644 --- a/sdk/cosmos/azure-cosmos-encryption/README.md +++ b/sdk/cosmos/azure-cosmos-encryption/README.md @@ -43,10 +43,10 @@ A few important properties are defined at the level of the container, among them ## Examples The following section provides several code snippets covering some of the most common Cosmos Encryption API tasks, including: -* [Create Cosmos Encryption Client](#create-cosmos-encryption-client "Create Cosmos Encryption Client") -* [Create Cosmos Encryption Database](#create-cosmos-encryption-database "Create Encryption Database") -* [Create Encryption Container](#create-cosmos-encryption-container "Create Encryption Container") -* [CRUD operation on Items](#crud-operation-on-items "CRUD operation on Items") +* [Create Cosmos Encryption Client](#create-cosmos-encryption-client) +* [Create Cosmos Encryption Database](#create-cosmos-encryption-database) +* [Create Encryption Container](#create-cosmos-encryption-container) +* [CRUD operation on Items](#crud-operation-on-items) ### Create Cosmos Encryption Client diff --git a/sdk/cosmos/azure-cosmos-test/README.md b/sdk/cosmos/azure-cosmos-test/README.md index ee9b439ab493..8ef815423527 100644 --- a/sdk/cosmos/azure-cosmos-test/README.md +++ b/sdk/cosmos/azure-cosmos-test/README.md @@ -32,10 +32,10 @@ The Azure Cosmos Test library can be used to inject failure into Azure Cosmos SD ## Examples The following section provides several code snippets covering how to create some of the most common failure injection scenario, including: -* [High Channel Acquisition Scenario](#high-channel-acquisition-scenario "High channel acquisition scenario") -* [Broken Connection Scenario](#broken-connection-scenario "Broken connection scenario") -* [Server Return Gone Scenario](#server-return-gone-scenario "Server gone scenario") -* [Random Connection Close Scenario](#random-connection-close-scenario "Random connection close scenario") +* [High Channel Acquisition Scenario](#high-channel-acquisition-scenario) +* [Broken Connection Scenario](#broken-connection-scenario) +* [Server Return Gone Scenario](#server-return-gone-scenario) +* [Random Connection Close Scenario](#random-connection-close-scenario) ### High Channel Acquisition Scenario