Don't call onComplete after onError in ChecksumValidatingSubscriber#onComplete method which results in NPE

Author: varunnvs92

.changes/next-release/bugfix-AWSSDKforJavav2-8adcc2e.json

@@ -0,0 +1,5 @@
+{
+    "category": "AWS SDK for Java v2", 
+    "type": "bugfix", 
+    "description": "Fix bug in ChecksumValidatingSubscriber which results in NPE if checksum validation fails."
+}

services/s3/src/main/java/software/amazon/awssdk/services/s3/checksums/ChecksumValidatingPublisher.java

@@ -138,6 +138,7 @@ public void onComplete() {
                     onError(SdkClientException.create(
                         String.format("Data read has a different checksum than expected. Was %d, but expected %d",
                                       computedChecksumInt, streamChecksumInt)));
+                    return; // Return after onError and not call onComplete below
                 }
             }
             wrapped.onComplete();

services/s3/src/test/java/software/amazon/awssdk/services/s3/checksums/ChecksumValidatingPublisherTest.java

@@ -17,20 +17,19 @@
 
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
-
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
-
 import software.amazon.awssdk.core.checksums.Md5Checksum;
 
 /**
@@ -66,6 +65,7 @@ public void testSinglePacket() {
     driver.doOnComplete();
 
     assertTrue(s.hasCompleted());
+    assertFalse(s.isOnErrorCalled());
   }
 
   @Test
@@ -81,6 +81,7 @@ public void testTwoPackets() {
       driver.doOnComplete();
 
       assertTrue(s.hasCompleted());
+      assertFalse(s.isOnErrorCalled());
     }
   }
 
@@ -100,6 +101,7 @@ public void testTinyPackets() {
       driver.doOnComplete();
 
       assertTrue(s.hasCompleted());
+      assertFalse(s.isOnErrorCalled());
     }
   }
 
@@ -121,12 +123,29 @@ public void testUnknownLength() {
     driver.doOnComplete();
 
     assertTrue(s.hasCompleted());
+    assertFalse(s.isOnErrorCalled());
+  }
+
+  @Test
+  public void checksumValidationFailure_throwsSdkClientException_NotNPE() {
+    final byte[] incorrectData = new byte[0];
+    final TestPublisher driver = new TestPublisher();
+    final TestSubscriber s = new TestSubscriber(Arrays.copyOfRange(incorrectData, 0, TEST_DATA_SIZE));
+    final ChecksumValidatingPublisher p = new ChecksumValidatingPublisher(driver, new Md5Checksum(), TEST_DATA_SIZE + CHECKSUM_SIZE);
+    p.subscribe(s);
+
+    driver.doOnNext(ByteBuffer.wrap(incorrectData));
+    driver.doOnComplete();
+
+    assertTrue(s.isOnErrorCalled());
+    assertFalse(s.hasCompleted());
   }
 
   private class TestSubscriber implements Subscriber<ByteBuffer> {
     final byte[] expected;
     final List<ByteBuffer> received;
     boolean completed;
+    boolean onErrorCalled;
 
     TestSubscriber(byte[] expected) {
       this.expected = expected;
@@ -148,10 +167,9 @@ public void onNext(ByteBuffer buffer) {
 
     @Override
     public void onError(Throwable t) {
-      fail("Test failed");
+      onErrorCalled = true;
     }
 
-
     @Override
     public void onComplete() {
       int matchPos = 0;
@@ -168,6 +186,10 @@ public void onComplete() {
     public boolean hasCompleted() {
       return completed;
     }
+
+    public boolean isOnErrorCalled() {
+      return onErrorCalled;
+    }
   }
 
   private class TestPublisher implements Publisher<ByteBuffer> {