Fix checkpointOwner copy issue for multistream lease (#1401)

Author: chenylee-aws

amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/Lease.java

@@ -163,6 +163,7 @@ protected Lease(Lease lease) {
                 lease.childShardIds(),
                 lease.pendingCheckpointState(),
                 lease.hashKeyRangeForLease());
+        checkpointOwner(lease.checkpointOwner);
     }
 
     @Deprecated
@@ -458,8 +459,6 @@ public void leaseOwner(String leaseOwner) {
      * @return A deep copy of this object.
      */
     public Lease copy() {
-        final Lease lease = new Lease(this);
-        lease.checkpointOwner(this.checkpointOwner);
-        return lease;
+        return new Lease(this);
     }
 }

amazon-kinesis-client/src/test/java/software/amazon/kinesis/leases/LeaseTest.java

@@ -10,6 +10,7 @@
 import org.mockito.runners.MockitoJUnitRunner;
 import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
@@ -109,6 +110,15 @@ public void testIsEligibleForGracefulShutdownFalse_shutdownRequested_assertFalse
         assertFalse(shutdownRequestedLease.isEligibleForGracefulShutdown());
     }
 
+    @Test
+    public void testCopyingLease() {
+        final String checkpointOwner = "checkpointOwner";
+        final Lease original = new Lease();
+        original.checkpointOwner(checkpointOwner);
+        final Lease copy = original.copy();
+        assertEquals(checkpointOwner, copy.checkpointOwner());
+    }
+
     private static Lease createLease(String leaseOwner, String leaseKey, long lastCounterIncrementNanos) {
         final Lease lease = new Lease();
         lease.checkpoint(new ExtendedSequenceNumber("checkpoint"));

amazon-kinesis-client/src/test/java/software/amazon/kinesis/leases/MultiStreamLeaseTest.java

@@ -0,0 +1,19 @@
+package software.amazon.kinesis.leases;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class MultiStreamLeaseTest {
+
+    @Test
+    void testCopyingMultiStreamLease() {
+        final String checkpointOwner = "checkpointOwner";
+        final MultiStreamLease original = new MultiStreamLease();
+        original.checkpointOwner(checkpointOwner);
+        original.streamIdentifier("identifier");
+        original.shardId("shardId");
+        final MultiStreamLease copy = original.copy();
+        assertEquals(checkpointOwner, copy.checkpointOwner());
+    }
+}