Skip to content

Fixed a CDK issue and Entity Resolution tests run again #7605

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Oct 2, 2025
Merged

Fixed a CDK issue and Entity Resolution tests run again #7605

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
207 changes: 119 additions & 88 deletions ...code/entityresolution/src/main/java/com/example/entity/scenario/CloudFormationHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,11 +10,7 @@
import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
import software.amazon.awssdk.services.cloudformation.CloudFormationAsyncClient;
import software.amazon.awssdk.services.cloudformation.model.Capability;
import software.amazon.awssdk.services.cloudformation.model.CloudFormationException;
import software.amazon.awssdk.services.cloudformation.model.DescribeStacksRequest;
import software.amazon.awssdk.services.cloudformation.model.DescribeStacksResponse;
import software.amazon.awssdk.services.cloudformation.model.Output;
import software.amazon.awssdk.services.cloudformation.model.*;
import software.amazon.awssdk.services.cloudformation.model.Stack;
import software.amazon.awssdk.services.cloudformation.waiters.CloudFormationAsyncWaiter;
import software.amazon.awssdk.services.s3.S3AsyncClient;
Expand All @@ -26,9 +22,7 @@
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

Expand All @@ -39,150 +33,187 @@ public class CloudFormationHelper {
private static CloudFormationAsyncClient cloudFormationClient;

public static void main(String[] args) {
if (args.length < 1) {
logger.error("Usage: CloudFormationHelper <bucketName>");
return;
}
emptyS3Bucket(args[0]);
}

private static CloudFormationAsyncClient getCloudFormationClient() {
if (cloudFormationClient == null) {
SdkAsyncHttpClient httpClient = NettyNioAsyncHttpClient.builder()
.maxConcurrency(100)
.connectionTimeout(Duration.ofSeconds(60))
.readTimeout(Duration.ofSeconds(60))
.writeTimeout(Duration.ofSeconds(60))
.build();
.maxConcurrency(100)
.connectionTimeout(Duration.ofSeconds(60))
.readTimeout(Duration.ofSeconds(60))
.writeTimeout(Duration.ofSeconds(60))
.build();

ClientOverrideConfiguration overrideConfig = ClientOverrideConfiguration.builder()
.apiCallTimeout(Duration.ofMinutes(2))
.apiCallAttemptTimeout(Duration.ofSeconds(90))
.retryStrategy(RetryMode.STANDARD)
.build();
.apiCallTimeout(Duration.ofMinutes(2))
.apiCallAttemptTimeout(Duration.ofSeconds(90))
.retryStrategy(RetryMode.STANDARD)
.build();

cloudFormationClient = CloudFormationAsyncClient.builder()
.httpClient(httpClient)
.overrideConfiguration(overrideConfig)
.build();
.httpClient(httpClient)
.overrideConfiguration(overrideConfig)
.build();
}
return cloudFormationClient;
}

public static void deployCloudFormationStack(String stackName) {
String templateBody;
logger.info("Deploying CloudFormation stack: {}", stackName);
boolean doesExist = describeStack(stackName);
if (!doesExist) {
String templateBody;
try {
ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
Path filePath = Paths.get(classLoader.getResource(CFN_TEMPLATE).toURI());
Path filePath = Paths.get(Objects.requireNonNull(classLoader.getResource(CFN_TEMPLATE)).toURI());
templateBody = Files.readString(filePath);
} catch (IOException | URISyntaxException e) {
logger.error("Failed to read CloudFormation template", e);
throw new RuntimeException(e);
}

getCloudFormationClient().createStack(b -> b.stackName(stackName)
.templateBody(templateBody)
.capabilities(Capability.CAPABILITY_IAM))
.whenComplete((csr, t) -> {
if (csr != null) {
System.out.println("Stack creation requested, ARN is " + csr.stackId());
.templateBody(templateBody)
.capabilities(Capability.CAPABILITY_IAM))
.whenComplete((csr, t) -> {
if (t != null) {
logger.error("Error creating stack {}", stackName, t);
throw new RuntimeException("Stack creation failed", t);
}

logger.info("Stack creation requested. ARN: {}", csr.stackId());

try (CloudFormationAsyncWaiter waiter = getCloudFormationClient().waiter()) {
waiter.waitUntilStackCreateComplete(request -> request.stackName(stackName))
.whenComplete((dsr, th) -> {
if (th != null) {
System.out.println("Error waiting for stack creation: " + th.getMessage());
} else {
dsr.matched().response().orElseThrow(() -> new RuntimeException("Failed to deploy"));
System.out.println("Stack created successfully");
}
}).join();
.whenComplete((dsr, th) -> {
if (th != null) {
logger.error("Error waiting for stack creation: {}", stackName, th);
} else {
dsr.matched().response()
.orElseThrow(() -> new RuntimeException("Stack creation failed for " + stackName));
logger.info("Stack {} created successfully.", stackName);

// Print outputs immediately
getStackOutputsAsync(stackName).whenComplete((outputs, outEx) -> {
if (outEx != null) {
logger.error("Failed to fetch stack outputs", outEx);
} else {
logger.info("Stack Outputs for {}:", stackName);
outputs.forEach((k, v) -> logger.info(" {} = {}", k, v));
}
}).join();
}
}).join();
}
} else {
System.out.format("Error creating stack: " + t.getMessage(), t);
throw new RuntimeException(t.getCause().getMessage(), t);
}
}).join();
}).join();
} else {
logger.info("{} stack already exists", stackName);
logger.info("Stack {} already exists, skipping creation.", stackName);
}
}

// Check to see if the Stack exists before deploying it
public static Boolean describeStack(String stackName) {
try {
CompletableFuture<?> future = getCloudFormationClient().describeStacks();
DescribeStacksResponse stacksResponse = (DescribeStacksResponse) future.join();
List<Stack> stacks = stacksResponse.stacks();
for (Stack myStack : stacks) {
if (myStack.stackName().compareTo(stackName) == 0) {
CompletableFuture<DescribeStacksResponse> future = getCloudFormationClient().describeStacks();
DescribeStacksResponse stacksResponse = future.join();
for (Stack myStack : stacksResponse.stacks()) {
if (myStack.stackName().equals(stackName)) {
logger.info("Stack {} exists already.", stackName);
return true;
}
}
} catch (CloudFormationException e) {
System.err.println(e.getMessage());
logger.error("Error describing stack {}", stackName, e);
}
return false;
}

public static void destroyCloudFormationStack(String stackName) {
logger.info("Deleting CloudFormation stack: {}", stackName);
getCloudFormationClient().deleteStack(b -> b.stackName(stackName))
.whenComplete((dsr, t) -> {
if (dsr != null) {
System.out.println("Delete stack requested ....");
.whenComplete((dsr, t) -> {
if (t != null) {
logger.error("Error deleting stack {}", stackName, t);
throw new RuntimeException("Delete failed", t);
}

logger.info("Delete stack requested: {}", stackName);
try (CloudFormationAsyncWaiter waiter = getCloudFormationClient().waiter()) {
waiter.waitUntilStackDeleteComplete(request -> request.stackName(stackName))
.whenComplete((waiterResponse, throwable) ->
System.out.println("Stack deleted successfully."))
.join();
.whenComplete((waiterResponse, throwable) -> {
if (throwable != null) {
logger.error("Error waiting for stack deletion {}", stackName, throwable);
} else {
logger.info("Stack {} deleted successfully.", stackName);
}
}).join();
}
} else {
System.out.format("Error deleting stack: " + t.getMessage(), t);
throw new RuntimeException(t.getCause().getMessage(), t);
}
}).join();
}).join();
}

public static CompletableFuture<Map<String, String>> getStackOutputsAsync(String stackName) {
CloudFormationAsyncClient cloudFormationAsyncClient = getCloudFormationClient();
logger.info("Fetching stack outputs for {}", stackName);

DescribeStacksRequest describeStacksRequest = DescribeStacksRequest.builder()
.stackName(stackName)
.build();

return cloudFormationAsyncClient.describeStacks(describeStacksRequest)
.handle((describeStacksResponse, throwable) -> {
if (throwable != null) {
throw new RuntimeException("Failed to get stack outputs for: " + stackName, throwable);
}
.stackName(stackName)
.build();

// Process the result
if (describeStacksResponse.stacks().isEmpty()) {
throw new RuntimeException("Stack not found: " + stackName);
}
return getCloudFormationClient().describeStacks(describeStacksRequest)
.handle((describeStacksResponse, throwable) -> {
if (throwable != null) {
logger.error("Failed to get stack outputs for {}", stackName, throwable);
throw new RuntimeException("Failed to get stack outputs for: " + stackName, throwable);
}

Stack stack = describeStacksResponse.stacks().get(0);
Map<String, String> outputs = new HashMap<>();
for (Output output : stack.outputs()) {
outputs.put(output.outputKey(), output.outputValue());
}
if (describeStacksResponse.stacks().isEmpty()) {
throw new RuntimeException("Stack not found: " + stackName);
}

return outputs;
});
Stack stack = describeStacksResponse.stacks().get(0);
Map<String, String> outputs = new HashMap<>();
for (Output output : stack.outputs()) {
outputs.put(output.outputKey(), output.outputValue());
}
return outputs;
});
}

public static void emptyS3Bucket(String bucketName) {
logger.info("Emptying S3 bucket: {}", bucketName);
S3AsyncClient s3Client = S3AsyncClient.builder().build();

s3Client.listObjectsV2(req -> req.bucket(bucketName))
.thenCompose(response -> {
List<CompletableFuture<DeleteObjectResponse>> deleteFutures = response.contents().stream()
.map(s3Object -> s3Client.deleteObject(req -> req
.bucket(bucketName)
.key(s3Object.key())))
.collect(Collectors.toList());
.thenCompose(response -> {
if (response.contents().isEmpty()) {
logger.info("Bucket {} is already empty.", bucketName);
return CompletableFuture.completedFuture(null);
}

return CompletableFuture.allOf(deleteFutures.toArray(new CompletableFuture[0]));
})
.join();
List<CompletableFuture<DeleteObjectResponse>> deleteFutures = response.contents().stream()
.map(s3Object -> {
logger.info("Deleting object: {}", s3Object.key());
return s3Client.deleteObject(req -> req
.bucket(bucketName)
.key(s3Object.key()));
})
.collect(Collectors.toList());

return CompletableFuture.allOf(deleteFutures.toArray(new CompletableFuture[0]));
})
.whenComplete((res, ex) -> {
if (ex != null) {
logger.error("Failed to empty bucket {}", bucketName, ex);
} else {
logger.info("Bucket {} emptied successfully.", bucketName);
}
})
.join();

s3Client.close();
}
}

2 changes: 1 addition & 1 deletion ...le_code/entityresolution/src/main/java/com/example/entity/scenario/EntityResScenario.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
public class EntityResScenario {
private static final Logger logger = LoggerFactory.getLogger(EntityResScenario.class);
public static final String DASHES = new String(new char[80]).replace("\0", "-");
private static final String STACK_NAME = "EntityResolutionCdkStack";
private static final String STACK_NAME = "EntityResolutionCdkStack2";
private static final String ENTITY_RESOLUTION_ROLE_ARN_KEY = "EntityResolutionRoleArn";
private static final String GLUE_DATA_BUCKET_NAME_KEY = "GlueDataBucketName";
private static final String JSON_GLUE_TABLE_ARN_KEY = "JsonErGlueTableArn";
Expand Down
Loading
Loading