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

Adding integration test for serializing application info

Browse source
This commit is contained in:
Joshua Kim 2020-04-07 06:16:04 -04:00
parent ddaf714a09
commit a61890ab40
1 changed files with 26 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
amazon-kinesis-client/src/test/java/software/amazon/kinesis/leases/dynamodb/DynamoDBLeaseCoordinatorIntegrationTest.java
View file

@ -127,16 +127,37 @@ public class DynamoDBLeaseCoordinatorIntegrationTest {
} }
assertNotNull(lease); assertNotNull(lease);
ExtendedSequenceNumber newCheckpoint = new ExtendedSequenceNumber("newCheckpoint"); final ExtendedSequenceNumber initialCheckpoint = new ExtendedSequenceNumber("initialCheckpoint");
final ExtendedSequenceNumber pendingCheckpoint = new ExtendedSequenceNumber("pendingCheckpoint");
final ExtendedSequenceNumber newCheckpoint = new ExtendedSequenceNumber("newCheckpoint");
final byte[] checkpointState = "checkpointState".getBytes();
// lease's leaseCounter is wrong at this point, but it shouldn't matter. // lease's leaseCounter is wrong at this point, but it shouldn't matter.
assertTrue(dynamoDBCheckpointer.setCheckpoint(lease.leaseKey(), initialCheckpoint, lease.concurrencyToken()));
final Lease leaseFromDDBAtInitialCheckpoint = leaseRefresher.getLease(lease.leaseKey());
lease.leaseCounter(lease.leaseCounter() + 1);
lease.checkpoint(initialCheckpoint);
lease.leaseOwner(coordinator.workerIdentifier());
assertEquals(lease, leaseFromDDBAtInitialCheckpoint);
dynamoDBCheckpointer.prepareCheckpoint(lease.leaseKey(), pendingCheckpoint, lease.concurrencyToken().toString(), checkpointState);
final Lease leaseFromDDBAtPendingCheckpoint = leaseRefresher.getLease(lease.leaseKey());
lease.leaseCounter(lease.leaseCounter() + 1);
lease.checkpoint(initialCheckpoint);
lease.pendingCheckpoint(pendingCheckpoint);
lease.pendingCheckpointState(checkpointState);
assertEquals(lease, leaseFromDDBAtPendingCheckpoint);
assertTrue(dynamoDBCheckpointer.setCheckpoint(lease.leaseKey(), newCheckpoint, lease.concurrencyToken())); assertTrue(dynamoDBCheckpointer.setCheckpoint(lease.leaseKey(), newCheckpoint, lease.concurrencyToken()));
Lease fromDynamo = leaseRefresher.getLease(lease.leaseKey()); final Lease leaseFromDDBAtNewCheckpoint = leaseRefresher.getLease(lease.leaseKey());
lease.leaseCounter(lease.leaseCounter() + 1); lease.leaseCounter(lease.leaseCounter() + 1);
lease.checkpoint(newCheckpoint); lease.checkpoint(newCheckpoint);
lease.leaseOwner(coordinator.workerIdentifier()); lease.pendingCheckpoint(null);
assertEquals(lease, fromDynamo); lease.pendingCheckpointState(null);
assertEquals(lease, leaseFromDDBAtNewCheckpoint);
} }
/** /**