DynamoDBIO { public static <T> Write<T> write() { return new AutoValue_DynamoDBIO_Write.Builder().build(); }  static Read<T> read(); static Write<T> write();  }
@Test public void testWriteDataToDynamo() { final List<WriteRequest> writeRequests = DynamoDBIOTestHelper.generateWriteRequests(numOfItems); final PCollection<Void> output = pipeline .apply(Create.of(writeRequests)) .apply( DynamoDBIO.<WriteRequest>write() .withWriteRequestMapperFn( (SerializableFunction<WriteRequest, KV<String, WriteRequest>>) writeRequest -> KV.of(tableName, writeRequest)) .withRetryConfiguration( DynamoDBIO.RetryConfiguration.create(5, Duration.standardMinutes(1))) .withAwsClientsProvider( AwsClientsProviderMock.of(DynamoDBIOTestHelper.getDynamoDBClient()))); final PCollection<Long> publishedResultsSize = output.apply(Count.globally()); PAssert.that(publishedResultsSize).containsInAnyOrder(0L); pipeline.run().waitUntilFinish(); }
@Test public void testRetries() throws Throwable { thrown.expectMessage("Error writing to DynamoDB"); final List<WriteRequest> writeRequests = DynamoDBIOTestHelper.generateWriteRequests(numOfItems); AmazonDynamoDB amazonDynamoDBMock = Mockito.mock(AmazonDynamoDB.class); Mockito.when(amazonDynamoDBMock.batchWriteItem(Mockito.any(BatchWriteItemRequest.class))) .thenThrow(new AmazonDynamoDBException("Service unavailable")); pipeline .apply(Create.of(writeRequests)) .apply( DynamoDBIO.<WriteRequest>write() .withWriteRequestMapperFn( (SerializableFunction<WriteRequest, KV<String, WriteRequest>>) writeRequest -> KV.of(tableName, writeRequest)) .withRetryConfiguration( DynamoDBIO.RetryConfiguration.create(4, Duration.standardSeconds(10))) .withAwsClientsProvider(AwsClientsProviderMock.of(amazonDynamoDBMock))); try { pipeline.run().waitUntilFinish(); } catch (final Pipeline.PipelineExecutionException e) { expectedLogs.verifyWarn(String.format(DynamoDBIO.Write.WriteFn.RETRY_ATTEMPT_LOG, 1)); expectedLogs.verifyWarn(String.format(DynamoDBIO.Write.WriteFn.RETRY_ATTEMPT_LOG, 2)); expectedLogs.verifyWarn(String.format(DynamoDBIO.Write.WriteFn.RETRY_ATTEMPT_LOG, 3)); throw e.getCause(); } fail("Pipeline is expected to fail because we were unable to write to DynamoDB."); }
@Test public void testWriteDataToDynamo() { List<KV<String, Integer>> items = ImmutableList.of(KV.of("test1", 111), KV.of("test2", 222), KV.of("test3", 333)); final PCollection<Void> output = pipeline .apply(Create.of(items)) .apply( DynamoDBIO.<KV<String, Integer>>write() .withWriteRequestMapperFn( (SerializableFunction<KV<String, Integer>, KV<String, WriteRequest>>) entry -> { Map<String, AttributeValue> putRequest = ImmutableMap.of( "hashKey1", AttributeValue.builder().s(entry.getKey()).build(), "rangeKey2", AttributeValue.builder() .n(entry.getValue().toString()) .build()); WriteRequest writeRequest = WriteRequest.builder() .putRequest(PutRequest.builder().item(putRequest).build()) .build(); return KV.of(tableName, writeRequest); }) .withRetryConfiguration( DynamoDBIO.RetryConfiguration.builder() .setMaxAttempts(5) .setMaxDuration(Duration.standardMinutes(1)) .setRetryPredicate(DEFAULT_RETRY_PREDICATE) .build()) .withDynamoDbClientProvider( DynamoDbClientProviderMock.of(DynamoDBIOTestHelper.getDynamoDBClient()))); final PCollection<Long> publishedResultsSize = output.apply(Count.globally()); PAssert.that(publishedResultsSize).containsInAnyOrder(0L); pipeline.run().waitUntilFinish(); int actualItemCount = DynamoDBIOTestHelper.readDataFromTable(tableName).size(); assertEquals(3, actualItemCount); }
@Test public void testRetries() throws Throwable { thrown.expectMessage("Error writing to DynamoDB"); List<KV<String, Integer>> items = ImmutableList.of(KV.of("test1", 111), KV.of("test2", 222), KV.of("test3", 333)); DynamoDbClient amazonDynamoDBMock = Mockito.mock(DynamoDbClient.class); Mockito.when(amazonDynamoDBMock.batchWriteItem(Mockito.any(BatchWriteItemRequest.class))) .thenThrow(DynamoDbException.builder().message("Service unavailable").build()); pipeline .apply(Create.of(items)) .apply( DynamoDBIO.<KV<String, Integer>>write() .withWriteRequestMapperFn( (SerializableFunction<KV<String, Integer>, KV<String, WriteRequest>>) entry -> { Map<String, AttributeValue> putRequest = ImmutableMap.of( "hashKey1", AttributeValue.builder().s(entry.getKey()).build(), "rangeKey2", AttributeValue.builder() .n(entry.getValue().toString()) .build()); WriteRequest writeRequest = WriteRequest.builder() .putRequest(PutRequest.builder().item(putRequest).build()) .build(); return KV.of(tableName, writeRequest); }) .withRetryConfiguration( DynamoDBIO.RetryConfiguration.builder() .setMaxAttempts(4) .setMaxDuration(Duration.standardSeconds(10)) .setRetryPredicate(DEFAULT_RETRY_PREDICATE) .build()) .withDynamoDbClientProvider(DynamoDbClientProviderMock.of(amazonDynamoDBMock))); try { pipeline.run().waitUntilFinish(); } catch (final Pipeline.PipelineExecutionException e) { expectedLogs.verifyWarn(String.format(DynamoDBIO.Write.WriteFn.RETRY_ATTEMPT_LOG, 1)); expectedLogs.verifyWarn(String.format(DynamoDBIO.Write.WriteFn.RETRY_ATTEMPT_LOG, 2)); expectedLogs.verifyWarn(String.format(DynamoDBIO.Write.WriteFn.RETRY_ATTEMPT_LOG, 3)); throw e.getCause(); } fail("Pipeline is expected to fail because we were unable to write to DynamoDb."); }