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

Address comments

Browse source
This commit is contained in:
Chunxue Yang 2020-04-22 15:00:10 -07:00
parent 93dac82bd6
commit cde09ca191
3 changed files with 18 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/dynamodb/DynamoDBLeaseSerializer.java
View file

@ -53,7 +53,7 @@ public class DynamoDBLeaseSerializer implements LeaseSerializer {
private static final String PENDING_CHECKPOINT_SUBSEQUENCE_KEY = "pendingCheckpointSubSequenceNumber"; private static final String PENDING_CHECKPOINT_SUBSEQUENCE_KEY = "pendingCheckpointSubSequenceNumber";
private static final String PENDING_CHECKPOINT_STATE_KEY = "pendingCheckpointState"; private static final String PENDING_CHECKPOINT_STATE_KEY = "pendingCheckpointState";
private static final String PARENT_SHARD_ID_KEY = "parentShardId"; private static final String PARENT_SHARD_ID_KEY = "parentShardId";
private static final String CHILD_SHARD_ID_KEY = "childShardIds"; private static final String CHILD_SHARD_IDS_KEY = "childShardIds";
@Override @Override
public Map<String, AttributeValue> toDynamoRecord(final Lease lease) { public Map<String, AttributeValue> toDynamoRecord(final Lease lease) {
@ -73,7 +73,7 @@ public class DynamoDBLeaseSerializer implements LeaseSerializer {
result.put(PARENT_SHARD_ID_KEY, DynamoUtils.createAttributeValue(lease.parentShardIds())); result.put(PARENT_SHARD_ID_KEY, DynamoUtils.createAttributeValue(lease.parentShardIds()));
} }
if (!CollectionUtils.isNullOrEmpty(lease.childShardIds())) { if (!CollectionUtils.isNullOrEmpty(lease.childShardIds())) {
result.put(CHILD_SHARD_ID_KEY, DynamoUtils.createAttributeValue(lease.childShardIds())); result.put(CHILD_SHARD_IDS_KEY, DynamoUtils.createAttributeValue(lease.childShardIds()));
} }
if (lease.pendingCheckpoint() != null && !lease.pendingCheckpoint().sequenceNumber().isEmpty()) { if (lease.pendingCheckpoint() != null && !lease.pendingCheckpoint().sequenceNumber().isEmpty()) {
@ -107,7 +107,7 @@ public class DynamoDBLeaseSerializer implements LeaseSerializer {
DynamoUtils.safeGetLong(dynamoRecord, CHECKPOINT_SUBSEQUENCE_NUMBER_KEY)) DynamoUtils.safeGetLong(dynamoRecord, CHECKPOINT_SUBSEQUENCE_NUMBER_KEY))
); );
leaseToUpdate.parentShardIds(DynamoUtils.safeGetSS(dynamoRecord, PARENT_SHARD_ID_KEY)); leaseToUpdate.parentShardIds(DynamoUtils.safeGetSS(dynamoRecord, PARENT_SHARD_ID_KEY));
leaseToUpdate.childShardIds(DynamoUtils.safeGetSS(dynamoRecord, CHILD_SHARD_ID_KEY)); leaseToUpdate.childShardIds(DynamoUtils.safeGetSS(dynamoRecord, CHILD_SHARD_IDS_KEY));
if (!Strings.isNullOrEmpty(DynamoUtils.safeGetString(dynamoRecord, PENDING_CHECKPOINT_SEQUENCE_KEY))) { if (!Strings.isNullOrEmpty(DynamoUtils.safeGetString(dynamoRecord, PENDING_CHECKPOINT_SEQUENCE_KEY))) {
leaseToUpdate.pendingCheckpoint( leaseToUpdate.pendingCheckpoint(
@ -243,7 +243,7 @@ public class DynamoDBLeaseSerializer implements LeaseSerializer {
if (!CollectionUtils.isNullOrEmpty(lease.childShardIds())) { if (!CollectionUtils.isNullOrEmpty(lease.childShardIds())) {
result.put(CHILD_SHARD_ID_KEY, putUpdate(DynamoUtils.createAttributeValue(lease.childShardIds()))); result.put(CHILD_SHARD_IDS_KEY, putUpdate(DynamoUtils.createAttributeValue(lease.childShardIds())));
} }
return result; return result;

12
amazon-kinesis-client/src/main/java/software/amazon/kinesis/lifecycle/ShutdownTask.java
View file

@ -52,6 +52,7 @@ import java.util.List;
import java.util.Set; import java.util.Set;
import java.util.UUID; import java.util.UUID;
import java.util.function.Function; import java.util.function.Function;
import java.util.stream.Collectors;
/** /**
* Task for invoking the ShardRecordProcessor shutdown() callback. * Task for invoking the ShardRecordProcessor shutdown() callback.
@ -195,14 +196,11 @@ public class ShutdownTask implements ConsumerTask {
private void updateLeasesForChildShards() private void updateLeasesForChildShards()
throws DependencyException, InvalidStateException, ProvisionedThroughputException { throws DependencyException, InvalidStateException, ProvisionedThroughputException {
final Lease currentLease = leaseCoordinator.getCurrentlyHeldLease(shardInfoIdProvider.apply(shardInfo)); final Lease currentLease = leaseCoordinator.getCurrentlyHeldLease(shardInfoIdProvider.apply(shardInfo));
Set<String> childShardIds = new HashSet<>(); Set<String> childShardIds = childShards.stream().map(ChildShard::shardId).collect(Collectors.toSet());
for (ChildShard childShard : childShards) {
childShardIds.add(childShard.shardId());
}
final Lease upatedLease = currentLease.copy(); final Lease updatedLease = currentLease.copy();
upatedLease.childShardIds(childShardIds); updatedLease.childShardIds(childShardIds);
leaseCoordinator.updateLease(upatedLease, UUID.fromString(shardInfo.concurrencyToken()), SHUTDOWN_TASK_OPERATION, shardInfoIdProvider.apply(shardInfo)); leaseCoordinator.updateLease(updatedLease, UUID.fromString(shardInfo.concurrencyToken()), SHUTDOWN_TASK_OPERATION, shardInfoIdProvider.apply(shardInfo));
} }
/* /*

15
amazon-kinesis-client/src/test/java/software/amazon/kinesis/lifecycle/ShutdownTaskTest.java
View file

@ -28,6 +28,7 @@ import static org.mockito.Mockito.when;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Arrays; import java.util.Arrays;
import java.util.Collection;
import java.util.Collections; import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.UUID; import java.util.UUID;
@ -123,7 +124,8 @@ public class ShutdownTaskTest {
@Test @Test
public final void testCallWhenApplicationDoesNotCheckpoint() { public final void testCallWhenApplicationDoesNotCheckpoint() {
when(recordProcessorCheckpointer.lastCheckpointValue()).thenReturn(new ExtendedSequenceNumber("3298")); when(recordProcessorCheckpointer.lastCheckpointValue()).thenReturn(new ExtendedSequenceNumber("3298"));
when(leaseCoordinator.getCurrentlyHeldLease("shardId-0")).thenReturn(createLease()); Lease heldLease = createLease("shardId-0", "leaseOwner", Collections.singleton("parentShardId"));
when(leaseCoordinator.getCurrentlyHeldLease("shardId-0")).thenReturn(heldLease);
when(leaseCoordinator.leaseRefresher()).thenReturn(leaseRefresher); when(leaseCoordinator.leaseRefresher()).thenReturn(leaseRefresher);
final TaskResult result = task.call(); final TaskResult result = task.call();
@ -164,7 +166,8 @@ public class ShutdownTaskTest {
hierarchicalShardSyncer, NULL_METRICS_FACTORY, constructChildShards()); hierarchicalShardSyncer, NULL_METRICS_FACTORY, constructChildShards());
when(recordProcessorCheckpointer.lastCheckpointValue()).thenReturn(ExtendedSequenceNumber.SHARD_END); when(recordProcessorCheckpointer.lastCheckpointValue()).thenReturn(ExtendedSequenceNumber.SHARD_END);
when(leaseCoordinator.getCurrentlyHeldLease("shardId-0")).thenReturn(createLease()); Lease heldLease = createLease("shardId-0", "leaseOwner", Collections.singleton("parentShardId"));
when(leaseCoordinator.getCurrentlyHeldLease("shardId-0")).thenReturn(heldLease);
when(leaseCoordinator.leaseRefresher()).thenReturn(leaseRefresher); when(leaseCoordinator.leaseRefresher()).thenReturn(leaseRefresher);
final TaskResult result = task.call(); final TaskResult result = task.call();
@ -253,11 +256,11 @@ public class ShutdownTaskTest {
return childShards; return childShards;
} }
private Lease createLease() { private Lease createLease(String leaseKey, String leaseOwner, Collection<String> parentShardIds) {
Lease lease = new Lease(); Lease lease = new Lease();
lease.leaseKey("shardId-0"); lease.leaseKey(leaseKey);
lease.leaseOwner("leaseOwner"); lease.leaseOwner(leaseOwner);
lease.parentShardIds(Collections.singleton("parentShardIds")); lease.parentShardIds(parentShardIds);
return lease; return lease;
} }