Skip to content

Commit fcaff26

Browse files
peterzxu-crmpeterzxu-crm
committed
W-17312191: Upgrade AWS SDK v2
1 parent 2c00a5e commit fcaff26

28 files changed

+456
-1127
lines changed

‎carbonj.service/build.gradle‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -201,8 +201,12 @@ dependencies {
201201
implementation group: 'io.netty', name: 'netty-all', version: "${nettyAll}"
202202
implementation group: 'net.razorvine', name: 'pickle', version: "${pickle}"
203203
implementation group: 'org.python', name: 'jython-standalone', version: "${jythonStandalone}"
204-
implementation group: 'com.amazonaws', name: 'amazon-kinesis-client', version: "${amazonKinesisClient}"
205-
implementation group: 'com.amazonaws', name: 'aws-java-sdk-sts', version: "${awsJavaSdkV1}"
204+
implementation group: 'org.apache.httpcomponents', name: 'httpclient', version: "${httpClient}"
205+
implementation group: 'software.amazon.kinesis', name: 'amazon-kinesis-client', version: "${awsKinesisClient}"
206+
implementation group: 'software.amazon.awssdk', name: 'kinesis', version: "${awsJavaSdkV2}"
207+
implementation group: 'software.amazon.awssdk', name: 'sts', version: "${awsJavaSdkV2}"
208+
implementation group: 'software.amazon.awssdk', name: 'dynamodb', version: "${awsJavaSdkV2}"
209+
implementation group: 'software.amazon.awssdk', name: 'netty-nio-client', version: "${awsJavaSdkV2}"
206210
implementation group: 'io.dropwizard.metrics', name: 'metrics-core', version: "${metrics}"
207211
implementation group: 'io.dropwizard.metrics', name: 'metrics-jvm', version: "${metrics}"
208212
implementation group: 'io.dropwizard.metrics', name: 'metrics-graphite', version: "${metrics}"

‎carbonj.service/src/main/java/com/demandware/carbonj/service/engine/AWSCredential.java‎

Lines changed: 0 additions & 38 deletions
This file was deleted.

‎carbonj.service/src/main/java/com/demandware/carbonj/service/engine/Consumers.java‎

Lines changed: 30 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,14 @@
1212
import com.demandware.carbonj.service.ns.NamespaceCounter;
1313
import org.slf4j.Logger;
1414
import org.slf4j.LoggerFactory;
15+
import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient;
16+
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
17+
import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
18+
import software.amazon.kinesis.common.ConfigsBuilder;
19+
import software.amazon.kinesis.common.InitialPositionInStream;
20+
import software.amazon.kinesis.common.InitialPositionInStreamExtended;
21+
import software.amazon.kinesis.processor.SingleStreamTracker;
22+
import software.amazon.kinesis.processor.StreamTracker;
1523

1624
import java.io.File;
1725
import java.io.FileInputStream;
@@ -25,6 +33,7 @@
2533
import java.util.Map;
2634
import java.util.Properties;
2735
import java.util.Set;
36+
import java.util.UUID;
2837
import java.util.concurrent.ConcurrentHashMap;
2938

3039
public class Consumers {
@@ -50,9 +59,18 @@ public class Consumers {
5059

5160
private final File indexNameSyncDir;
5261

62+
private final String activeProfile;
63+
64+
private final KinesisAsyncClient kinesisAsyncClient;
65+
66+
private final DynamoDbAsyncClient dynamoDbAsyncClient;
67+
68+
private final CloudWatchAsyncClient cloudWatchAsyncClient;
69+
5370
Consumers(MetricRegistry metricRegistry, PointProcessor pointProcessor, PointProcessor recoveryPointProcessor, File rulesFile,
5471
KinesisConfig kinesisConfig, CheckPointMgr<Date> checkPointMgr, String kinesisConsumerRegion,
55-
NamespaceCounter namespaceCounter, File indexNameSyncDir) {
72+
NamespaceCounter namespaceCounter, File indexNameSyncDir, String activeProfile,
73+
KinesisAsyncClient kinesisAsyncClient, DynamoDbAsyncClient dynamoDbAsyncClient, CloudWatchAsyncClient cloudWatchAsyncClient) {
5674

5775
this.metricRegistry = metricRegistry;
5876
this.pointProcessor = pointProcessor;
@@ -62,6 +80,10 @@ public class Consumers {
6280
this.kinesisConsumerRegion = kinesisConsumerRegion;
6381
this.namespaceCounter = namespaceCounter;
6482
this.indexNameSyncDir = indexNameSyncDir;
83+
this.activeProfile = activeProfile;
84+
this.kinesisAsyncClient = kinesisAsyncClient;
85+
this.dynamoDbAsyncClient = dynamoDbAsyncClient;
86+
this.cloudWatchAsyncClient = cloudWatchAsyncClient;
6587
consumers = new ConcurrentHashMap<>();
6688
consumerRules = new ConsumerRules(rulesFile);
6789
reload();
@@ -139,8 +161,12 @@ private void reconfigureConsumers(Set<String> newRules, Set<String> currentRules
139161
}
140162

141163
Counter initRetryCounter = metricRegistry.counter(MetricRegistry.name("kinesis.consumer." + kinesisStreamName + ".initRetryCounter"));
142-
KinesisConsumer kinesisConsumer = new KinesisConsumer(metricRegistry, pointProcessor, recoveryPointProcessor, kinesisStreamName,
143-
kinesisApplicationName, kinesisConfig, checkPointMgr, initRetryCounter, kinesisConsumerRegion);
164+
StreamTracker streamTracker = new SingleStreamTracker(kinesisStreamName,
165+
InitialPositionInStreamExtended.newInitialPosition(InitialPositionInStream.TRIM_HORIZON));
166+
ConfigsBuilder configsBuilder = new ConfigsBuilder(streamTracker, kinesisApplicationName, kinesisAsyncClient,
167+
dynamoDbAsyncClient, cloudWatchAsyncClient, UUID.randomUUID().toString(),
168+
new KinesisRecordProcessorFactory(metricRegistry, pointProcessor, kinesisConfig, kinesisStreamName));
169+
KinesisConsumer kinesisConsumer = new KinesisConsumer(kinesisStreamName, kinesisApplicationName, initRetryCounter, configsBuilder);
144170
log.info(String.format("New Consumer created with name %s", kinesisStreamName));
145171
newConsumers.add(consumerName);
146172
consumers.put(consumerName, kinesisConsumer);
@@ -171,7 +197,7 @@ private String getHostName() {
171197
}
172198

173199
private String getKinesisApplicationName(String streamName, String hostName) {
174-
return streamName + "-" + hostName;
200+
return streamName + "-" + hostName + "-" + activeProfile;
175201
}
176202

177203
private void close(Set<String> consumerSet) {

‎carbonj.service/src/main/java/com/demandware/carbonj/service/engine/DynamoDbCheckPointMgr.java‎

Lines changed: 52 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -6,23 +6,25 @@
66
*/
77
package com.demandware.carbonj.service.engine;
88

9-
import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
10-
import com.amazonaws.services.dynamodbv2.document.DynamoDB;
11-
import com.amazonaws.services.dynamodbv2.document.Table;
12-
import com.amazonaws.services.dynamodbv2.model.AttributeDefinition;
13-
import com.amazonaws.services.dynamodbv2.model.AttributeValue;
14-
import com.amazonaws.services.dynamodbv2.model.CreateTableRequest;
15-
import com.amazonaws.services.dynamodbv2.model.GetItemRequest;
16-
import com.amazonaws.services.dynamodbv2.model.KeySchemaElement;
17-
import com.amazonaws.services.dynamodbv2.model.KeyType;
18-
import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughput;
19-
import com.amazonaws.services.dynamodbv2.model.ScalarAttributeType;
209
import org.slf4j.Logger;
2110
import org.slf4j.LoggerFactory;
11+
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
12+
import software.amazon.awssdk.services.dynamodb.model.AttributeDefinition;
13+
import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
14+
import software.amazon.awssdk.services.dynamodb.model.CreateTableRequest;
15+
import software.amazon.awssdk.services.dynamodb.model.CreateTableResponse;
16+
import software.amazon.awssdk.services.dynamodb.model.GetItemRequest;
17+
import software.amazon.awssdk.services.dynamodb.model.GetItemResponse;
18+
import software.amazon.awssdk.services.dynamodb.model.KeySchemaElement;
19+
import software.amazon.awssdk.services.dynamodb.model.KeyType;
20+
import software.amazon.awssdk.services.dynamodb.model.ProvisionedThroughput;
21+
import software.amazon.awssdk.services.dynamodb.model.ScalarAttributeType;
22+
import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest;
2223

2324
import java.util.Date;
2425
import java.util.HashMap;
2526
import java.util.Map;
27+
import java.util.concurrent.CompletableFuture;
2628
import java.util.concurrent.TimeUnit;
2729

2830
public class DynamoDbCheckPointMgr implements CheckPointMgr<Date> {
@@ -31,71 +33,74 @@ public class DynamoDbCheckPointMgr implements CheckPointMgr<Date> {
3133
private final String tableName;
3234
private final int defaultOffsetMins;
3335

34-
private final AmazonDynamoDB client;
35-
private final DynamoDB dynamoDB;
36+
private final DynamoDbAsyncClient client;
37+
private final int checkPointDynamodbTimout;
3638

37-
public DynamoDbCheckPointMgr(AmazonDynamoDB client, String kinesisApplicationName, int defaultOffsetMins,
38-
int provisionedThroughput) throws Exception {
39+
public DynamoDbCheckPointMgr(DynamoDbAsyncClient client, String kinesisApplicationName, int defaultOffsetMins,
40+
int provisionedThroughput, int checkPointDynamodbTimout) throws Exception {
3941
this.client = client;
40-
this.dynamoDB = new DynamoDB(client);
4142
this.defaultOffsetMins = defaultOffsetMins;
4243
this.tableName = "checkpoints-" + kinesisApplicationName;
43-
if (!DynamoDbUtils.isTablePresent(dynamoDB, tableName)) {
44+
this.checkPointDynamodbTimout = checkPointDynamodbTimout;
45+
if (!DynamoDbUtils.isTablePresent(client, tableName, checkPointDynamodbTimout)) {
4446
createTable(tableName, provisionedThroughput);
4547
}
4648
}
4749

4850
private void createTable(String tableName, int provisionedThroughput) throws Exception {
49-
CreateTableRequest request = new CreateTableRequest()
50-
.withAttributeDefinitions(
51-
new AttributeDefinition("checkPointType", ScalarAttributeType.S))
52-
.withKeySchema(
53-
new KeySchemaElement("checkPointType", KeyType.HASH))
54-
.withProvisionedThroughput(
55-
new ProvisionedThroughput((long)provisionedThroughput, (long)provisionedThroughput))
56-
.withTableName(tableName);
51+
CreateTableRequest request = CreateTableRequest.builder()
52+
.tableName(tableName)
53+
.attributeDefinitions(AttributeDefinition.builder().attributeName("checkPointType").attributeType(ScalarAttributeType.S).build())
54+
.keySchema(KeySchemaElement.builder().attributeName("checkPointType").keyType(KeyType.HASH).build())
55+
.provisionedThroughput(ProvisionedThroughput.builder()
56+
.readCapacityUnits((long)provisionedThroughput)
57+
.writeCapacityUnits((long)provisionedThroughput)
58+
.build())
59+
.build();
5760
log.info("Issuing CreateTable request for " + tableName);
58-
Table newlyCreatedTable = dynamoDB.createTable(request);
61+
CompletableFuture<CreateTableResponse> createTableResponse = this.client.createTable(request);
5962
log.info("Waiting for " + tableName + " to be created...this may take a while...");
60-
newlyCreatedTable.waitForActive();
63+
createTableResponse.get(checkPointDynamodbTimout, TimeUnit.SECONDS);
6164
}
6265

6366
@Override
6467
public void checkPoint(Date checkPoint) throws Exception {
65-
Table table = dynamoDB.getTable(tableName);
6668

67-
HashMap<String, String> expressionAttributeNames = new HashMap<>();
69+
Map<String, String> expressionAttributeNames = new HashMap<>();
6870
expressionAttributeNames.put("#V", "checkPointValue");
6971

70-
HashMap<String, Object> expressionAttributeValues = new HashMap<>();
71-
expressionAttributeValues.put(":val1", checkPoint.getTime());
72+
Map<String, AttributeValue> expressionAttributeValues = new HashMap<>();
73+
expressionAttributeValues.put(":val1", AttributeValue.builder().n(String.valueOf(checkPoint.getTime())).build());
7274

73-
table.updateItem(
74-
"checkPointType", // key attribute name
75-
"timestamp", // key attribute value
76-
"set #V = :val1", // UpdateExpression
77-
expressionAttributeNames,
78-
expressionAttributeValues);
75+
client.updateItem(UpdateItemRequest.builder()
76+
.tableName(tableName)
77+
.key(Map.of("checkPointType", AttributeValue.builder().s("timestamp").build()))
78+
.updateExpression("set #V = :val1")
79+
.expressionAttributeNames(expressionAttributeNames)
80+
.expressionAttributeValues(expressionAttributeValues).build());
7981
}
8082

8183
@Override
8284
public Date lastCheckPoint() throws Exception {
83-
HashMap<String, AttributeValue> keyToGet = new HashMap<String, AttributeValue>();
84-
keyToGet.put( "checkPointType", new AttributeValue( "timestamp") );
85-
GetItemRequest request = new GetItemRequest()
86-
.withKey( keyToGet )
87-
.withTableName( tableName );
88-
89-
Map<String, AttributeValue> item = client.getItem( request ).getItem();
90-
if( item == null ) {
85+
86+
GetItemRequest request = GetItemRequest.builder()
87+
.tableName(tableName)
88+
.key(Map.of("checkPointType", AttributeValue.builder().s("timestamp").build()))
89+
.build();
90+
91+
GetItemResponse getItemResponse = this.client.getItem(request).get(checkPointDynamodbTimout, TimeUnit.SECONDS);
92+
93+
if (!getItemResponse.hasItem()) {
9194
return getDefaultCheckPoint();
9295
}
93-
String value = item.get( "checkPointValue" ).getN();
96+
97+
Map<String, AttributeValue> item = getItemResponse.item();
98+
String value = item.get("checkPointValue").n();
9499
if( value == null ) {
95100
return getDefaultCheckPoint();
96101
}
97102

98-
return new Date( Long.parseLong( value ) );
103+
return new Date(Long.parseLong(value));
99104
}
100105

101106
private Date getDefaultCheckPoint() {
@@ -104,4 +109,3 @@ private Date getDefaultCheckPoint() {
104109
return checkPoint;
105110
}
106111
}
107-

‎carbonj.service/src/main/java/com/demandware/carbonj/service/engine/DynamoDbUtils.java‎

Lines changed: 16 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -6,35 +6,33 @@
66
*/
77
package com.demandware.carbonj.service.engine;
88

9-
import com.amazonaws.services.dynamodbv2.document.DynamoDB;
10-
import com.amazonaws.services.dynamodbv2.document.Table;
11-
import com.amazonaws.services.dynamodbv2.model.ResourceNotFoundException;
12-
import com.amazonaws.services.dynamodbv2.model.TableDescription;
139
import org.slf4j.Logger;
1410
import org.slf4j.LoggerFactory;
11+
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
12+
import software.amazon.awssdk.services.dynamodb.model.DescribeTableRequest;
13+
import software.amazon.awssdk.services.dynamodb.model.DescribeTableResponse;
14+
import software.amazon.awssdk.services.dynamodb.model.TableDescription;
15+
import software.amazon.awssdk.services.dynamodb.model.TableStatus;
16+
17+
import java.util.concurrent.ExecutionException;
18+
import java.util.concurrent.TimeUnit;
19+
import java.util.concurrent.TimeoutException;
1520

1621
public class DynamoDbUtils {
1722

1823
private static final Logger log = LoggerFactory.getLogger(DynamoDbUtils.class);
1924

20-
public static boolean isTablePresent(DynamoDB dynamoDB, String tableName) {
21-
Table table = dynamoDB.getTable(tableName);
25+
public static boolean isTablePresent(DynamoDbAsyncClient client, String tableName, int checkPointDynamodbTimout) throws InterruptedException {
26+
DescribeTableRequest describeTableRequest = DescribeTableRequest.builder().tableName(tableName).build();
2227

2328
try {
24-
TableDescription tableDescription = table.describe();
25-
return "ACTIVE".equals(tableDescription.getTableStatus());
26-
} catch (ResourceNotFoundException e) {
29+
DescribeTableResponse describeTableResponse = client.describeTable(describeTableRequest)
30+
.get(checkPointDynamodbTimout, TimeUnit.SECONDS);
31+
TableDescription tableDescription = describeTableResponse.table();
32+
return tableDescription.tableStatus() == TableStatus.ACTIVE;
33+
} catch (ExecutionException | TimeoutException e) {
2734
log.warn("kinesis consumer table '" + tableName + "' not found!");
2835
return false;
2936
}
3037
}
31-
32-
public static void deleteTable(DynamoDB dynamoDB, String tableName) throws InterruptedException {
33-
Table table = dynamoDB.getTable(tableName);
34-
table.delete();
35-
36-
log.info("Waiting for " + tableName + " to be deleted...this may take a while...");
37-
38-
table.waitForDelete();
39-
}
4038
}

0 commit comments

Comments
 (0)