[Kafka Connector]Fix BadRequest exception when using serverless database account

sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md

@@ -7,6 +7,7 @@
 #### Breaking Changes
 
 #### Bugs Fixed
+* Fixed `BadRequestException` when customer using Serverless CosmosDB database account and metadata container does not exists. - See [PR 43125](https://github.com/Azure/azure-sdk-for-java/pull/43125) 
 
 #### Other Changes
 

sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/CosmosSourceConnector.java

@@ -215,12 +215,28 @@ private boolean shouldCreateMetadataContainerIfNotExists() {
     }
 
     private Mono<CosmosContainerResponse> createMetadataContainer() {
+        return this.createMetadataContainer(METADATA_CONTAINER_DEFAULT_RU_CONFIG)
+            .onErrorResume(throwable -> {
+                // for serverless database account, creating container with throughput configured will get 400/0 exceptions
+                if (KafkaCosmosExceptionsHelper.isBadRequestException(throwable)) {
+                    LOGGER.info(
+                        "Getting exception '{}' when creating metadata container with throughput, retrying creating without throughput.",
+                        throwable.getMessage());
+                    return this.createMetadataContainer(null);
+                }
+
+                return Mono.error(throwable);
+            });
+    }
+
+    private Mono<CosmosContainerResponse> createMetadataContainer(Integer throughput) {
         return this.cosmosClient
             .getDatabase(this.config.getContainersConfig().getDatabaseName())
             .createContainer(
                 this.config.getMetadataConfig().getStorageName(),
                 "/id",
-                ThroughputProperties.createAutoscaledThroughput(METADATA_CONTAINER_DEFAULT_RU_CONFIG));
+                throughput == null
+                    ? null : ThroughputProperties.createAutoscaledThroughput(throughput));
     }
 
     private void updateMetadataRecordsInCosmos(MetadataTaskUnit metadataTaskUnit) {

sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/KafkaCosmosExceptionsHelper.java

@@ -94,4 +94,11 @@ public static boolean isOwnerResourceNotExistsException(Throwable throwable) {
             && cosmosException.getStatusCode() == HttpConstants.StatusCodes.NOTFOUND
             && cosmosException.getSubStatusCode() == HttpConstants.SubStatusCodes.OWNER_RESOURCE_NOT_EXISTS;
     }
+
+    public static boolean isBadRequestException(Throwable throwable) {
+        CosmosException cosmosException = Utils.as(Exceptions.unwrap(throwable), CosmosException.class);
+        return cosmosException != null
+            && cosmosException.getStatusCode() == HttpConstants.StatusCodes.BADREQUEST
+            && cosmosException.getSubStatusCode() == HttpConstants.SubStatusCodes.UNKNOWN;
+    }
 }

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

@@ -50,8 +50,8 @@ public class CosmosSourceConnectorITest extends KafkaCosmosIntegrationTestSuiteB
     @BeforeClass(groups = { "kafka-integration" })
     public void before_CosmosSourceConnectorITest() {
         this.client = new CosmosClientBuilder()
-            .key(TestConfigurations.MASTER_KEY)
-            .endpoint(TestConfigurations.HOST)
+            .key(KafkaCosmosTestConfigurations.MASTER_KEY)
+            .endpoint(KafkaCosmosTestConfigurations.HOST)
             .endpointDiscoveryEnabled(true)
             .buildAsyncClient();
     }