Fixing encryption create with contentResponseOnWriteEnabled false (Azure#23218)

* Fixing encryption create with contentResponseOnWriteEnabled false

* resolving comments

* fixing spot bugs

Author: simplynaveen20

sdk/cosmos/azure-cosmos-encryption/src/main/java/com/azure/cosmos/encryption/CosmosEncryptionAsyncContainer.java

@@ -183,6 +183,10 @@ public <T> Mono<CosmosItemResponse<T>> readItem(String id,
                                                     PartitionKey partitionKey,
                                                     CosmosItemRequestOptions requestOptions,
                                                     Class<T> classType) {
+        if (requestOptions == null) {
+            requestOptions = new CosmosItemRequestOptions();
+        }
+
         Mono<CosmosItemResponse<byte[]>> responseMessageMono = this.readItemHelper(id, partitionKey, requestOptions, false);
 
         return responseMessageMono.publishOn(encryptionScheduler).flatMap(cosmosItemResponse -> setByteArrayContent(cosmosItemResponse,
@@ -311,7 +315,7 @@ private Mono<CosmosItemResponse<byte[]>> setByteArrayContent(CosmosItemResponse<
                 this.cosmosItemResponseBuilderAccessor.setByteArrayContent(rsp, bytes);
                 return Mono.just(rsp);
             }
-        );
+        ).defaultIfEmpty(rsp);
     }
 
     private <T> Function<CosmosPagedFluxOptions, Flux<FeedResponse<T>>> queryDecryptionTransformer(Class<T> classType,

sdk/cosmos/azure-cosmos-encryption/src/main/java/com/azure/cosmos/encryption/implementation/EncryptionProcessor.java

@@ -344,8 +344,12 @@ public Mono<byte[]> decrypt(byte[] input) {
                 input == null ? null : input.length,
                 Thread.currentThread().getName());
         }
-        ObjectNode itemJObj = Utils.parse(input, ObjectNode.class);
 
+        if (input == null || input.length == 0) {
+            return Mono.empty();
+        }
+
+        ObjectNode itemJObj = Utils.parse(input, ObjectNode.class);
         assert (itemJObj != null);
         return initEncryptionSettingsIfNotInitializedAsync().then(Mono.defer(() -> {
             for (ClientEncryptionIncludedPath includedPath : this.clientEncryptionPolicy.getIncludedPaths()) {

sdk/cosmos/azure-cosmos-encryption/src/test/java/com/azure/cosmos/encryption/EncryptionAsyncApiCrudTest.java

@@ -60,7 +60,7 @@ public EncryptionAsyncApiCrudTest(CosmosClientBuilder clientBuilder) {
     public void before_CosmosItemTest() {
         assertThat(this.client).isNull();
         this.client = getClientBuilder().buildAsyncClient();
-        EncryptionKeyStoreProvider encryptionKeyStoreProvider = new TestEncryptionKeyStoreProvider();
+        TestEncryptionKeyStoreProvider encryptionKeyStoreProvider = new TestEncryptionKeyStoreProvider();
         cosmosAsyncDatabase = getSharedCosmosDatabase(this.client);
         cosmosEncryptionAsyncClient = CosmosEncryptionAsyncClient.createCosmosEncryptionAsyncClient(this.client,
             encryptionKeyStoreProvider);
@@ -123,6 +123,17 @@ public void createItemEncrypt_readItemDecrypt() {
         validateResponse(properties, responseItem);
     }
 
+    @Test(groups = {"encryption"}, timeOut = TIMEOUT)
+    public void createItemEncryptWithContentResponseOnWriteEnabledFalse() {
+        CosmosItemRequestOptions requestOptions = new CosmosItemRequestOptions();
+        requestOptions.setContentResponseOnWriteEnabled(false);
+        EncryptionPojo properties = getItem(UUID.randomUUID().toString());
+        CosmosItemResponse<EncryptionPojo> itemResponse = cosmosEncryptionAsyncContainer.createItem(properties,
+            new PartitionKey(properties.getMypk()), requestOptions).block();
+        assertThat(itemResponse.getRequestCharge()).isGreaterThan(0);
+        assertThat(itemResponse.getItem()).isNull();
+    }
+
     @Test(groups = {"encryption"}, timeOut = TIMEOUT)
     public void upsertItem_readItem() {
         EncryptionPojo properties = getItem(UUID.randomUUID().toString());