public static String setBucketNotificationToEventBridge(String bucketName, String topicArn, String queueArn) {
        try {
            // Enable bucket to emit S3 Event notifications to EventBridge.
            s3Client.putBucketNotificationConfiguration(b -> b
                    .bucket(bucketName)
                    .notificationConfiguration(b1 -> b1
                            .eventBridgeConfiguration(
                                    SdkBuilder::build)
                    ).build()).join();

            // Create an EventBridge rule to route Object Created notifications.
            PutRuleRequest putRuleRequest = PutRuleRequest.builder()
                    .name(RULE_NAME)
                    .eventPattern("""
                            {
                              "source": ["aws.s3"],
                              "detail-type": ["Object Created"],
                              "detail": {
                                "bucket": {
                                  "name": ["%s"]
                                }
                              }
                            }
                            """.formatted(bucketName))
                    .build();

            // Add the rule to the default event bus.
            PutRuleResponse putRuleResponse = eventBridgeClient.putRule(putRuleRequest)
                    .whenComplete((r, t) -> {
                        if (t != null) {
                            logger.error("Error creating event bus rule: " + t.getMessage(), t);
                            throw new RuntimeException(t.getCause().getMessage(), t);
                        }
                        logger.info("Event bus rule creation request sent successfully. ARN is: {}", r.ruleArn());
                    }).join();

            // Add the existing SNS topic and SQS queue as targets to the rule.
            eventBridgeClient.putTargets(b -> b
                    .eventBusName("default")
                    .rule(RULE_NAME)
                    .targets(List.of (
                            Target.builder()
                                    .arn(queueArn)
                                    .id("Queue")
                                    .build(),
                            Target.builder()
                                    .arn(topicArn)
                                    .id("Topic")
                                    .build())
                            )
                    ).join();
            return putRuleResponse.ruleArn();
        } catch (S3Exception e) {
            System.err.println(e.awsErrorDetails().errorMessage());
            System.exit(1);
        }
        return null;
    }
@Test
    @Tag("IntegrationTest")
    void setBucketNotificationToEventBridge() {

        String ruleArn = PutBucketS3EventNotificationEventBridge.setBucketNotificationToEventBridge(bucketName, topicArn, directToQueueArn);
        PutBucketS3EventNotificationEventBridge.addPermissions(directToQueueArn, directToQueueUrl,
                subscriberQueueArn, subscriberQueueUrl, topicArn, ruleArn);
        triggerS3EventMessages();
        try {
            Thread.sleep(Duration.ofSeconds(30).toMillis()); // Wait for messages to route through.
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        List<String> urls = getQueueUrls();
        urls.forEach(url -> {
                    ReceiveMessageResponse response = sqsClient.receiveMessage(b -> b
                            .queueUrl(url)).join();
                    logger.info("Messages received at queue {}: {}", url, response);
                    assertTrue(response.hasMessages());
                }
        );
    }