Addressing half of the PR comments. Have questions on the rest

Author: Michael Li

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/batchmanager/BatchBuffer.java

@@ -58,21 +58,21 @@ public BatchBuffer(ScheduledFuture<?> scheduledFlush) {
         this.scheduledFlush = scheduledFlush;
     }
 
-    public Map<String, BatchingExecutionContext<RequestT, ResponseT>> canManualFlush(int maxBatchItems) {
+    public Map<String, BatchingExecutionContext<RequestT, ResponseT>> flushableRequests(int maxBatchItems) {
         synchronized (flushLock) {
             if (idToBatchContext.size() >= maxBatchItems) {
                 return extractFlushedEntries(maxBatchItems);
             }
-            return null;
+            return new ConcurrentHashMap<>();
         }
     }
 
-    public Map<String, BatchingExecutionContext<RequestT, ResponseT>> canScheduledFlush(int maxBatchItems) {
+    public Map<String, BatchingExecutionContext<RequestT, ResponseT>> flushableScheduledRequests(int maxBatchItems) {
         synchronized (flushLock) {
             if (idToBatchContext.size() > 0) {
                 return extractFlushedEntries(maxBatchItems);
             }
-            return null;
+            return new ConcurrentHashMap<>();
         }
     }
 

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/batchmanager/BatchManager.java

@@ -107,18 +107,21 @@ public CompletableFuture<ResponseT> sendRequest(RequestT request) {
                                          () -> scheduleBufferFlush(batchKey, maxBatchOpenInMs.toMillis(), scheduledExecutor),
                                          request,
                                          response);
-
-            Map<String, BatchingExecutionContext<RequestT, ResponseT>> flushableRequests =
-                requestsAndResponsesMaps.canManualFlush(batchKey, maxBatchItems);
-            if (flushableRequests != null) {
-                manualFlushBuffer(batchKey, flushableRequests);
-            }
+            flushBufferIfNeeded(batchKey);
         } catch (Exception e) {
             response.completeExceptionally(e);
         }
         return response;
     }
 
+    private void flushBufferIfNeeded(String batchKey) {
+        Map<String, BatchingExecutionContext<RequestT, ResponseT>> flushableRequests =
+            requestsAndResponsesMaps.flushableRequests(batchKey, maxBatchItems);
+        if (!flushableRequests.isEmpty()) {
+            manualFlushBuffer(batchKey, flushableRequests);
+        }
+    }
+
     private void manualFlushBuffer(String batchKey,
                                    Map<String, BatchingExecutionContext<RequestT, ResponseT>> flushableRequests) {
         requestsAndResponsesMaps.cancelScheduledFlush(batchKey);
@@ -168,20 +171,21 @@ private ScheduledFuture<?> scheduleBufferFlush(String batchKey, long timeOutInMs
     }
 
     private void performScheduledFlush(String batchKey) {
-        Map<String, BatchingExecutionContext<RequestT, ResponseT>> requests =
-            requestsAndResponsesMaps.canScheduledFlush(batchKey, maxBatchItems);
-        if (requests != null) {
-            flushBuffer(batchKey, requests);
+        Map<String, BatchingExecutionContext<RequestT, ResponseT>> flushableRequests =
+            requestsAndResponsesMaps.flushableScheduledRequests(batchKey, maxBatchItems);
+        if (!flushableRequests.isEmpty()) {
+            flushBuffer(batchKey, flushableRequests);
         }
     }
 
     public void close() {
         requestsAndResponsesMaps.forEach((batchKey, batchBuffer) -> {
             requestsAndResponsesMaps.cancelScheduledFlush(batchKey);
+            Map<String, BatchingExecutionContext<RequestT, ResponseT>> flushableRequests =
+                requestsAndResponsesMaps.flushableScheduledRequests(batchKey, maxBatchItems);
 
-            Map<String, BatchingExecutionContext<RequestT, ResponseT>> requests;
-            while ((requests = requestsAndResponsesMaps.canScheduledFlush(batchKey, maxBatchItems)) != null) {
-                flushBuffer(batchKey, requests);
+            while (!flushableRequests.isEmpty()) {
+                flushBuffer(batchKey, flushableRequests);
             }
         });
         requestsAndResponsesMaps.waitForFlushesAndClear(log);

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/batchmanager/BatchUtils.java

@@ -25,15 +25,12 @@ private BatchUtils() {
     }
 
     public static String getAndIncrementId(AtomicInteger id) {
-        int currentId;
-        int newCurrentId;
-        do {
-            currentId = id.get();
-            newCurrentId = currentId + 1;
-            if (newCurrentId == Integer.MAX_VALUE) {
-                newCurrentId = 0;
-            }
-        } while (!id.compareAndSet(currentId, newCurrentId));
+        int currentId = id.get();
+        int newCurrentId = currentId + 1;
+        if (newCurrentId == Integer.MAX_VALUE) {
+            newCurrentId = 0;
+        }
+        id.set(newCurrentId);
         return Integer.toString(currentId);
     }
 }

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/batchmanager/BatchingMap.java

@@ -54,14 +54,14 @@ public void forEach(BiConsumer<String, BatchBuffer<RequestT, ResponseT>> action)
         batchContextMap.forEach(action);
     }
 
-    public Map<String, BatchingExecutionContext<RequestT, ResponseT>> canManualFlush(String batchKey,
-                                                                                               int maxBatchItems) {
-        return batchContextMap.get(batchKey).canManualFlush(maxBatchItems);
+    public Map<String, BatchingExecutionContext<RequestT, ResponseT>> flushableRequests(String batchKey,
+                                                                                        int maxBatchItems) {
+        return batchContextMap.get(batchKey).flushableRequests(maxBatchItems);
     }
 
-    public Map<String, BatchingExecutionContext<RequestT, ResponseT>> canScheduledFlush(String batchKey,
-                                                                                                  int maxBatchItems) {
-        return batchContextMap.get(batchKey).canScheduledFlush(maxBatchItems);
+    public Map<String, BatchingExecutionContext<RequestT, ResponseT>> flushableScheduledRequests(String batchKey,
+                                                                                                 int maxBatchItems) {
+        return batchContextMap.get(batchKey).flushableScheduledRequests(maxBatchItems);
     }
 
     public void cancelScheduledFlush(String batchKey) {

core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/batchmanager/IdentifiableMessageTest.java

@@ -36,15 +36,15 @@ public void checkIdenticalIdentifiableMessagesAreEqual() {
         IdentifiableMessage<String> myRequest1 = new IdentifiableMessage<>(id, request);
         IdentifiableMessage<String> myRequest2 = new IdentifiableMessage<>(id, request);
         Assert.assertEquals(myRequest1, myRequest2);
+        Assert.assertEquals(myRequest1.hashCode(), myRequest2.hashCode());
     }
 
     @Test
-    public void identifiableMessageHashCode() {
-        String id = "id";
-        String request = "request";
-        IdentifiableMessage<String> myRequest1 = new IdentifiableMessage<>(id, request);
-        IdentifiableMessage<String> myRequest2 = new IdentifiableMessage<>(id, request);
-        Assert.assertEquals(myRequest1.hashCode(), myRequest2.hashCode());
+    public void checkIdenticalIdentifiableMessagesAreNotEqual() {
+        IdentifiableMessage<String> myRequest1 = new IdentifiableMessage<>("id1", "request1");
+        IdentifiableMessage<String> myRequest2 = new IdentifiableMessage<>("id2", "request2");
+        Assert.assertNotEquals(myRequest1, myRequest2);
+        Assert.assertNotEquals(myRequest1.hashCode(), myRequest2.hashCode());
     }
 
 }

services/sqs/src/it/java/software/amazon/awssdk/services/sqs/BatchManagerSqsIntegrationTest.java

@@ -159,7 +159,6 @@ public void scheduleFiveMessagesWithEachThreadToDifferentLocations() {
         checkThreadedResponses(requests, responses, sendRequestFutures);
     }
 
-    // Sometimes it passes a null identifiedRequests;
     BatchAndSend<SendMessageRequest, SendMessageBatchResponse> batchingFunction =
         (identifiedRequests, destination) -> {
             List<SendMessageBatchRequestEntry> entries = new ArrayList<>(identifiedRequests.size());