updates to address Abhijeet review comments

Author: TheovanKraay

src/main/java/com/azure/cosmos/examples/bulk/async/BulkWriter.java

@@ -17,9 +17,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
 import reactor.core.scheduler.Schedulers;
 
+import java.time.Duration;
 import java.util.concurrent.Semaphore;
 
 public class BulkWriter {
@@ -111,21 +113,17 @@ private void processResponseCode(
                 "The operation for Item ID: [{}]  Item PartitionKey Value: [{}] will be retried",
                 itemOperation.getId(),
                 itemOperation.getPartitionKeyValue());
-            if (itemResponse.getRetryAfterDuration() != null) {
-                logger.info(
-                    "Retrying after [{}] milliseconds",
-                    itemResponse.getRetryAfterDuration().toMillis());
-                try {
-                    Thread.sleep(itemResponse.getRetryAfterDuration().toMillis());
-                } catch (InterruptedException e) {
-                    throw new RuntimeException(e);
-                }
-            } else {
-                logger.info("Retrying without delay");
-            }
+            long delay = itemResponse.getRetryAfterDuration() != null
+                    ? itemResponse.getRetryAfterDuration().toMillis()
+                    : 0;
+
+            logger.info("Retrying after [{}] ms", delay);
+
+            Mono.delay(Duration.ofMillis(delay))
+                    // re-scheduling
+                    .doOnNext(t -> scheduleWrites(itemOperation))
+                    .subscribe();
 
-            //re-scheduling
-            scheduleWrites(itemOperation);
         } else {
             logger.info(
                 "The operation for Item ID: [{}]  Item PartitionKey Value: [{}] did not complete successfully " +
@@ -156,7 +154,11 @@ private void handleException(CosmosItemOperation itemOperation, Exception except
     }
 
     private boolean shouldRetry(int statusCode) {
-        return statusCode == HttpConstants.StatusCodes.REQUEST_TIMEOUT ||
-            statusCode == HttpConstants.StatusCodes.TOO_MANY_REQUESTS;
+        return statusCode == 408 ||
+                statusCode == 429 ||
+                statusCode == 503 ||
+                statusCode == 500 ||
+                statusCode == 449 ||
+                statusCode == 410;
     }
 }

src/main/java/com/azure/cosmos/examples/bulk/sync/BulkWriter.java

@@ -19,6 +19,8 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Semaphore;
 
 public class BulkWriter {
@@ -27,7 +29,7 @@ public class BulkWriter {
     private final CosmosContainer cosmosContainer;
     private final int cpuCount = Runtime.getRuntime().availableProcessors();
     private final Semaphore semaphore = new Semaphore(1024 * 167 / cpuCount);
-    private final List<CosmosItemOperation> bufferedOperations = new ArrayList<>();
+    private final Queue<CosmosItemOperation> bufferedOperations = new ConcurrentLinkedQueue<>();
     private final int maxRetries = 5;
 
     public BulkWriter(CosmosContainer cosmosContainer) {
@@ -168,7 +170,11 @@ private void handleException(
     }
 
     private boolean shouldRetry(int statusCode) {
-        return statusCode == HttpConstants.StatusCodes.REQUEST_TIMEOUT ||
-                statusCode == HttpConstants.StatusCodes.TOO_MANY_REQUESTS;
+        return statusCode == 408 ||
+                statusCode == 429 ||
+                statusCode == 503 ||
+                statusCode == 500 ||
+                statusCode == 449 ||
+                statusCode == 410;
     }
 }