luciano/amazon-kinesis-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Adding method to lose leases. Fixing tests to do what the previous did.

Browse source
This commit is contained in:
Sahil Palvia 2017-07-12 13:49:04 -07:00
parent 45fa933bc0
commit 359b810601
3 changed files with 26 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
src/test/java/com/amazonaws/services/kinesis/leases/impl/LeaseIntegrationTest.java
View file

@ -16,6 +16,7 @@ package com.amazonaws.services.kinesis.leases.impl;
import java.util.logging.Logger; import java.util.logging.Logger;
import com.amazonaws.services.kinesis.leases.exceptions.LeasingException;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
import org.junit.Ignore; import org.junit.Ignore;
@ -71,4 +72,9 @@ public class LeaseIntegrationTest {
} }
}; };
protected void loseLease(KinesisClientLease lease) throws LeasingException {
// Need to call update 2 times because 1st time would be considered to be spurious update for the lease.
leaseManager.updateLease(lease);
leaseManager.updateLease(lease);
}
} }

3
src/test/java/com/amazonaws/services/kinesis/leases/impl/LeaseManagerIntegrationTest.java
View file

@ -108,9 +108,10 @@ public class LeaseManagerIntegrationTest extends LeaseIntegrationTest {
KinesisClientLease leaseCopy = leaseManager.getLease(lease.getLeaseKey()); KinesisClientLease leaseCopy = leaseManager.getLease(lease.getLeaseKey());
leaseManager.renewLease(lease);
leaseManager.renewLease(lease); leaseManager.renewLease(lease);
Assert.assertTrue(leaseManager.renewLease(leaseCopy)); Assert.assertFalse(leaseManager.renewLease(leaseCopy));
} }
/** /**

33
src/test/java/com/amazonaws/services/kinesis/leases/impl/LeaseRenewerIntegrationTest.java
View file

@ -57,8 +57,10 @@ public class LeaseRenewerIntegrationTest extends LeaseIntegrationTest {
builder.addLeasesToRenew(renewer, "1", "2"); builder.addLeasesToRenew(renewer, "1", "2");
KinesisClientLease renewedLease = builder.renewMutateAssert(renewer, "1", "2").get("2"); KinesisClientLease renewedLease = builder.renewMutateAssert(renewer, "1", "2").get("2");
leaseManager.updateLease(renewedLease); // lose lease 2
builder.renewMutateAssert(renewer, "1", "2"); loseLease(renewedLease);
builder.renewMutateAssert(renewer, "1");
} }
@Test @Test
@ -95,9 +97,9 @@ public class LeaseRenewerIntegrationTest extends LeaseIntegrationTest {
public void testGetCurrentlyHeldLeases() throws LeasingException { public void testGetCurrentlyHeldLeases() throws LeasingException {
TestHarnessBuilder builder = new TestHarnessBuilder(leaseManager); TestHarnessBuilder builder = new TestHarnessBuilder(leaseManager);
KinesisClientLease lease2 = builder.withLease("1", "foo").withLease("2", "foo").build().get("2"); builder.withLease("1", "foo").withLease("2", "foo").build();
builder.addLeasesToRenew(renewer, "1", "2"); builder.addLeasesToRenew(renewer, "1", "2");
builder.renewMutateAssert(renewer, "1", "2"); KinesisClientLease lease2 = builder.renewMutateAssert(renewer, "1", "2").get("2");
// This should be a copy that doesn't get updated // This should be a copy that doesn't get updated
Map<String, KinesisClientLease> heldLeases = renewer.getCurrentlyHeldLeases(); Map<String, KinesisClientLease> heldLeases = renewer.getCurrentlyHeldLeases();
@ -105,14 +107,15 @@ public class LeaseRenewerIntegrationTest extends LeaseIntegrationTest {
Assert.assertEquals((Long) 1L, heldLeases.get("1").getLeaseCounter()); Assert.assertEquals((Long) 1L, heldLeases.get("1").getLeaseCounter());
Assert.assertEquals((Long) 1L, heldLeases.get("2").getLeaseCounter()); Assert.assertEquals((Long) 1L, heldLeases.get("2").getLeaseCounter());
leaseManager.updateLease(lease2); // lose lease 2 // lose lease 2
// Do another renewal and make sure the copy doesn't change loseLease(lease2);
builder.renewMutateAssert(renewer, "1", "2");
// Do another renewal and make sure the copy doesn't change
builder.renewMutateAssert(renewer, "1");
heldLeases = renewer.getCurrentlyHeldLeases();
Assert.assertEquals(2, heldLeases.size()); Assert.assertEquals(2, heldLeases.size());
Assert.assertEquals((Long) 2L, heldLeases.get("1").getLeaseCounter()); Assert.assertEquals((Long) 1L, heldLeases.get("1").getLeaseCounter());
Assert.assertEquals((Long) 2L, heldLeases.get("2").getLeaseCounter()); Assert.assertEquals((Long) 1L, heldLeases.get("2").getLeaseCounter());
} }
@Test @Test
@ -176,11 +179,11 @@ public class LeaseRenewerIntegrationTest extends LeaseIntegrationTest {
KinesisClientLease lease = renewer.getCurrentlyHeldLease("1"); KinesisClientLease lease = renewer.getCurrentlyHeldLease("1");
// cause lease loss such that the renewer knows the lease has been lost when update is called // cause lease loss such that the renewer knows the lease has been lost when update is called
leaseManager.renewLease(lease); loseLease(lease);
builder.renewMutateAssert(renewer, "1"); builder.renewMutateAssert(renewer);
lease.setCheckpoint(new ExtendedSequenceNumber("new checkpoint")); lease.setCheckpoint(new ExtendedSequenceNumber("new checkpoint"));
Assert.assertTrue(renewer.updateLease(lease, lease.getConcurrencyToken())); Assert.assertFalse(renewer.updateLease(lease, lease.getConcurrencyToken()));
} }
@Test @Test
@ -195,8 +198,8 @@ public class LeaseRenewerIntegrationTest extends LeaseIntegrationTest {
KinesisClientLease lease = renewer.getCurrentlyHeldLease("1"); KinesisClientLease lease = renewer.getCurrentlyHeldLease("1");
// cause lease loss such that the renewer knows the lease has been lost when update is called // cause lease loss such that the renewer knows the lease has been lost when update is called
leaseManager.renewLease(lease); loseLease(lease);
builder.renewMutateAssert(renewer, "1"); builder.renewMutateAssert(renewer);
// regain the lease // regain the lease
builder.addLeasesToRenew(renewer, "1"); builder.addLeasesToRenew(renewer, "1");