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

Few minor revisions

Browse source
This commit is contained in:
Meher Mankikar 2023-06-19 09:37:09 -07:00
parent ae42e96e55
commit 97d7831446
4 changed files with 16 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/RecordValidatorQueue.java
View file

@ -26,7 +26,7 @@ public class RecordValidatorQueue {
values.add(data); values.add(data);
} }
public RecordValidationStatus validateRecords(int trueTotalShardCount) { public RecordValidationStatus validateRecords(int expectedShardCount) {
// Validate that each List in the HashMap has data records in increasing order // Validate that each List in the HashMap has data records in increasing order
boolean incOrder = true; boolean incOrder = true;
@ -54,16 +54,16 @@ public class RecordValidatorQueue {
} }
// Validate that no records are missing over all shards // Validate that no records are missing over all shards
int totalShardCount = 0; int actualShardCount = 0;
for (Map.Entry<String, List<String>> entry : dict.entrySet()) { for (Map.Entry<String, List<String>> entry : dict.entrySet()) {
List<String> recordsPerShard = entry.getValue(); List<String> recordsPerShard = entry.getValue();
Set<String> noDupRecords = new HashSet<String>(recordsPerShard); Set<String> noDupRecords = new HashSet<String>(recordsPerShard);
totalShardCount += noDupRecords.size(); actualShardCount += noDupRecords.size();
} }
// If this is true, then there was some record that was missed during processing. // If this is true, then there was some record that was missed during processing.
if (totalShardCount != trueTotalShardCount) { if (actualShardCount != expectedShardCount) {
log.error("Failed to get correct number of records processed. Should be {} but was {}", trueTotalShardCount, totalShardCount); log.error("Failed to get correct number of records processed. Should be {} but was {}", expectedShardCount, actualShardCount);
return RecordValidationStatus.MISSING_RECORD; return RecordValidationStatus.MISSING_RECORD;
} }

6
amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/StreamExistenceManager.java
View file

@ -43,12 +43,10 @@ public class StreamExistenceManager {
throw new RuntimeException("Stream is not active, instead in status: " + response.streamDescriptionSummary().streamStatus()); throw new RuntimeException("Stream is not active, instead in status: " + response.streamDescriptionSummary().streamStatus());
} }
return true; return true;
} catch (ExecutionException e) { } catch (ResourceNotFoundException e) {
if (e.getCause() instanceof ResourceNotFoundException) {
return false; return false;
} else { } catch (ExecutionException e) {
throw new RuntimeException(e); throw new RuntimeException(e);
}
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }

4
amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestConsumer.java
View file

@ -189,9 +189,11 @@ public class TestConsumer {
} }
private void deleteResources(StreamExistenceManager streamExistenceManager, DynamoDbAsyncClient dynamoDBClient) throws Exception { private void deleteResources(StreamExistenceManager streamExistenceManager, DynamoDbAsyncClient dynamoDBClient) throws Exception {
log.info("-------------Start deleting test resources.----------------"); log.info("-------------Start deleting stream.----------------");
streamExistenceManager.deleteStream(this.streamName); streamExistenceManager.deleteStream(this.streamName);
log.info("-------------Start deleting lease table.----------------");
deleteLeaseTable(dynamoDBClient, consumerConfig.getStreamName()); deleteLeaseTable(dynamoDBClient, consumerConfig.getStreamName());
log.info("-------------Finished deleting resources.----------------");
} }
private void deleteLeaseTable(DynamoDbAsyncClient dynamoClient, String tableName) throws Exception { private void deleteLeaseTable(DynamoDbAsyncClient dynamoClient, String tableName) throws Exception {

6
amazon-kinesis-client/src/test/java/software/amazon/kinesis/utils/TestRecordProcessor.java
View file

@ -5,6 +5,7 @@ import org.slf4j.MDC;
import software.amazon.kinesis.exceptions.InvalidStateException; import software.amazon.kinesis.exceptions.InvalidStateException;
import software.amazon.kinesis.exceptions.ShutdownException; import software.amazon.kinesis.exceptions.ShutdownException;
import software.amazon.kinesis.lifecycle.events.LeaseLostInput; import software.amazon.kinesis.lifecycle.events.LeaseLostInput;
import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput;
import software.amazon.kinesis.lifecycle.events.ShardEndedInput; import software.amazon.kinesis.lifecycle.events.ShardEndedInput;
import software.amazon.kinesis.lifecycle.events.ShutdownRequestedInput; import software.amazon.kinesis.lifecycle.events.ShutdownRequestedInput;
import software.amazon.kinesis.processor.ShardRecordProcessor; import software.amazon.kinesis.processor.ShardRecordProcessor;
@ -13,6 +14,9 @@ import software.amazon.kinesis.retrieval.KinesisClientRecord;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
/**
* Implement initialization and deletion of shards and shard record processing
*/
@Slf4j @Slf4j
public class TestRecordProcessor implements ShardRecordProcessor { public class TestRecordProcessor implements ShardRecordProcessor {
@ -39,7 +43,7 @@ public class TestRecordProcessor implements ShardRecordProcessor {
@Override @Override
public void processRecords(software.amazon.kinesis.lifecycle.events.ProcessRecordsInput processRecordsInput) { public void processRecords(ProcessRecordsInput processRecordsInput) {
MDC.put(SHARD_ID_MDC_KEY, shardId); MDC.put(SHARD_ID_MDC_KEY, shardId);
try { try {
log.info("Processing {} record(s)", processRecordsInput.records().size()); log.info("Processing {} record(s)", processRecordsInput.records().size());