This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
The following commit(s) were added to refs/heads/master by this push:
new 39e8b73 Switches the AWS v2 test infra-structure
39e8b73 is described below
commit 39e8b739d5721fab09588cd71a8e5fd17f976ad8
Author: Otavio Rodolfo Piske <opi...@redhat.com>
AuthorDate: Fri Oct 9 14:30:54 2020 +0200
Switches the AWS v2 test infra-structure
Replaces the current AWS v2 test services with the reusable ones that
were moved to Camel core
---
tests/itests-aws-v2/pom.xml | 14 ++-
.../aws/v2/clients/AWSSDKClientUtils.java | 108 ---------------------
.../aws/v2/common/TestAWSCredentialsProvider.java | 70 -------------
.../source/CamelAWSKinesisPropertyFactory.java | 2 +-
.../source/CamelSourceAWSKinesisITCase.java | 6 +-
.../kinesis/source/TestKinesisConfiguration.java | 2 +-
.../v2/s3/source/CamelAWSS3PropertyFactory.java | 2 +-
.../aws/v2/s3/source/CamelSourceAWSS3ITCase.java | 8 +-
.../aws/v2/s3/source/TestS3Configuration.java | 2 +-
.../aws/v2/services/AWSContainer.java | 79 ---------------
.../services/AWSKinesisLocalContainerService.java | 47 ---------
.../aws/v2/services/AWSLocalContainerService.java | 94 ------------------
.../aws/v2/services/AWSRemoteService.java | 67 -------------
.../v2/services/AWSS3LocalContainerService.java | 47 ---------
.../v2/services/AWSSQSLocalContainerService.java | 48 ---------
.../aws/v2/services/AWSServiceFactory.java | 95 ------------------
.../kafkaconnector/aws/v2/services/Service.java | 38 --------
.../v2/sqs/sink/CamelAWSSQSPropertyFactory.java | 2 +-
.../aws/v2/sqs/sink/CamelSinkAWSSQSITCase.java | 8 +-
.../v2/sqs/source/CamelAWSSQSPropertyFactory.java | 2 +-
.../aws/v2/sqs/source/CamelSourceAWSSQSITCase.java | 8 +-
21 files changed, 32 insertions(+), 717 deletions(-)
diff --git a/tests/itests-aws-v2/pom.xml b/tests/itests-aws-v2/pom.xml
index dd3b24d..d52e712 100644
--- a/tests/itests-aws-v2/pom.xml
+++ b/tests/itests-aws-v2/pom.xml
@@ -38,9 +38,17 @@
</dependency>
<dependency>
- <groupId>org.apache.camel.kafkaconnector</groupId>
- <artifactId>itests-aws-common</artifactId>
- <version>${project.version}</version>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-infra-aws-common</artifactId>
+ <version>${camel.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-infra-aws-v2</artifactId>
+ <version>${camel.version}</version>
<type>test-jar</type>
<scope>test</scope>
</dependency>
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/clients/AWSSDKClientUtils.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/clients/AWSSDKClientUtils.java
index 60e9e1e..5fd3339 100644
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/clients/AWSSDKClientUtils.java
+++
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/clients/AWSSDKClientUtils.java
@@ -17,126 +17,18 @@
package org.apache.camel.kafkaconnector.aws.v2.clients;
-import java.net.URI;
-import java.net.URISyntaxException;
-
-import org.apache.camel.kafkaconnector.aws.common.AWSConfigs;
-import
org.apache.camel.kafkaconnector.aws.v2.common.TestAWSCredentialsProvider;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import software.amazon.awssdk.regions.Region;
-import software.amazon.awssdk.services.kinesis.KinesisClient;
-import software.amazon.awssdk.services.kinesis.KinesisClientBuilder;
import software.amazon.awssdk.services.s3.S3Client;
-import software.amazon.awssdk.services.s3.S3ClientBuilder;
import software.amazon.awssdk.services.s3.model.DeleteBucketRequest;
import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
import software.amazon.awssdk.services.s3.model.S3Object;
-import software.amazon.awssdk.services.sqs.SqsClient;
-import software.amazon.awssdk.services.sqs.SqsClientBuilder;
public final class AWSSDKClientUtils {
- private static final Logger LOG =
LoggerFactory.getLogger(AWSSDKClientUtils.class);
-
private AWSSDKClientUtils() {
}
- private static URI getEndpoint() {
- String amazonHost = System.getProperty(AWSConfigs.AMAZON_AWS_HOST);
-
- if (amazonHost == null || amazonHost.isEmpty()) {
- return null;
- }
-
- try {
- return new URI(String.format("http://%s";, amazonHost));
- } catch (URISyntaxException e) {
- throw new RuntimeException("Invalid endpoint");
- }
- }
-
- private static boolean isLocalContainer(String awsInstanceType) {
- return awsInstanceType == null ||
awsInstanceType.equals("local-aws-container");
- }
-
- public static KinesisClient newKinesisClient() {
- LOG.debug("Creating a new AWS v2 Kinesis client");
-
- String awsInstanceType =
System.getProperty("aws-service.kinesis.instance.type");
-
- KinesisClientBuilder clientBuilder = KinesisClient.builder();
-
- clientBuilder.region(Region.US_EAST_1);
-
- URI endpoint = getEndpoint();
-
- if (isLocalContainer(awsInstanceType) || endpoint != null) {
- clientBuilder.endpointOverride(endpoint);
- }
-
- if (isLocalContainer(awsInstanceType)) {
-
clientBuilder.credentialsProvider(TestAWSCredentialsProvider.CONTAINER_LOCAL_DEFAULT_PROVIDER);
-
- } else {
-
clientBuilder.credentialsProvider(TestAWSCredentialsProvider.SYSTEM_PROPERTY_PROVIDER);
- }
-
- return clientBuilder.build();
- }
-
- public static SqsClient newSQSClient() {
- LOG.debug("Creating a new AWS v2 SQS client");
-
- String awsInstanceType =
System.getProperty("aws-service.instance.type");
-
- SqsClientBuilder clientBuilder = SqsClient.builder();
-
- clientBuilder.region(Region.US_EAST_1);
-
- URI endpoint = getEndpoint();
-
- if (isLocalContainer(awsInstanceType) || endpoint != null) {
- clientBuilder.endpointOverride(endpoint);
- }
-
- if (isLocalContainer(awsInstanceType)) {
-
clientBuilder.credentialsProvider(TestAWSCredentialsProvider.CONTAINER_LOCAL_DEFAULT_PROVIDER);
-
- } else {
-
clientBuilder.credentialsProvider(TestAWSCredentialsProvider.SYSTEM_PROPERTY_PROVIDER);
- }
-
- return clientBuilder.build();
- }
-
- public static S3Client newS3Client() {
- LOG.debug("Creating a new S3 client");
- S3ClientBuilder clientBuilder = S3Client.builder();
-
- String awsInstanceType =
System.getProperty("aws-service.instance.type");
-
- clientBuilder.region(Region.US_EAST_1);
-
- URI endpoint = getEndpoint();
-
- if (isLocalContainer(awsInstanceType) || endpoint != null) {
- clientBuilder.endpointOverride(endpoint);
- }
-
- if (isLocalContainer(awsInstanceType)) {
-
clientBuilder.credentialsProvider(TestAWSCredentialsProvider.CONTAINER_LOCAL_DEFAULT_PROVIDER);
-
- } else {
-
clientBuilder.credentialsProvider(TestAWSCredentialsProvider.SYSTEM_PROPERTY_PROVIDER);
- }
-
- return clientBuilder.build();
- }
-
-
/**
* Delete an S3 bucket using the provided client. Coming from AWS
documentation:
* https://docs.aws.amazon.com/AmazonS3/latest/dev/Versioning.html
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/common/TestAWSCredentialsProvider.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/common/TestAWSCredentialsProvider.java
deleted file mode 100644
index bad1a69..0000000
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/common/TestAWSCredentialsProvider.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.camel.kafkaconnector.aws.v2.common;
-
-import org.apache.camel.kafkaconnector.aws.common.AWSConfigs;
-import software.amazon.awssdk.auth.credentials.AwsCredentials;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-
-public class TestAWSCredentialsProvider implements AwsCredentialsProvider {
- public static final TestAWSCredentialsProvider
CONTAINER_LOCAL_DEFAULT_PROVIDER =
- new TestAWSCredentialsProvider("accesskey", "secretkey");
-
- public static final TestAWSCredentialsProvider SYSTEM_PROPERTY_PROVIDER =
new TestAWSCredentialsProvider();
-
- private static class TestAWSCredentials implements AwsCredentials {
- private final String accessKey;
- private final String secretKey;
-
-
- public TestAWSCredentials() {
- this(System.getProperty(AWSConfigs.ACCESS_KEY),
System.getProperty(AWSConfigs.SECRET_KEY));
- }
-
- public TestAWSCredentials(String accessKey, String secretKey) {
- this.accessKey = accessKey;
- this.secretKey = secretKey;
- }
-
- @Override
- public String accessKeyId() {
- return accessKey;
- }
-
- @Override
- public String secretAccessKey() {
- return secretKey;
- }
-
- };
-
- private AwsCredentials credentials;
-
- public TestAWSCredentialsProvider() {
- credentials = new TestAWSCredentials();
- }
-
-
- public TestAWSCredentialsProvider(String accessKey, String secretKey) {
- credentials = new TestAWSCredentials(accessKey, secretKey);
- }
-
- @Override
- public AwsCredentials resolveCredentials() {
- return credentials;
- }
-}
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/kinesis/source/CamelAWSKinesisPropertyFactory.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/kinesis/source/CamelAWSKinesisPropertyFactory.java
index a021b1c..27e8fc3 100644
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/kinesis/source/CamelAWSKinesisPropertyFactory.java
+++
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/kinesis/source/CamelAWSKinesisPropertyFactory.java
@@ -21,9 +21,9 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
-import org.apache.camel.kafkaconnector.aws.common.AWSConfigs;
import org.apache.camel.kafkaconnector.common.EndpointUrlBuilder;
import org.apache.camel.kafkaconnector.common.SourceConnectorPropertyFactory;
+import org.apache.camel.test.infra.aws.common.AWSConfigs;
import software.amazon.awssdk.regions.Region;
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/kinesis/source/CamelSourceAWSKinesisITCase.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/kinesis/source/CamelSourceAWSKinesisITCase.java
index 33b8792..b5cca54 100644
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/kinesis/source/CamelSourceAWSKinesisITCase.java
+++
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/kinesis/source/CamelSourceAWSKinesisITCase.java
@@ -22,13 +22,13 @@ import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-import org.apache.camel.kafkaconnector.aws.common.AWSCommon;
-import org.apache.camel.kafkaconnector.aws.common.services.AWSService;
-import org.apache.camel.kafkaconnector.aws.v2.services.AWSServiceFactory;
import org.apache.camel.kafkaconnector.common.AbstractKafkaTest;
import org.apache.camel.kafkaconnector.common.ConnectorPropertyFactory;
import org.apache.camel.kafkaconnector.common.clients.kafka.KafkaClient;
import org.apache.camel.kafkaconnector.common.utils.TestUtils;
+import org.apache.camel.test.infra.aws.common.AWSCommon;
+import org.apache.camel.test.infra.aws.common.services.AWSService;
+import org.apache.camel.test.infra.aws2.services.AWSServiceFactory;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/kinesis/source/TestKinesisConfiguration.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/kinesis/source/TestKinesisConfiguration.java
index 6f8906f..20ef00a 100644
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/kinesis/source/TestKinesisConfiguration.java
+++
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/kinesis/source/TestKinesisConfiguration.java
@@ -18,7 +18,7 @@
package org.apache.camel.kafkaconnector.aws.v2.kinesis.source;
import org.apache.camel.component.aws2.kinesis.Kinesis2Configuration;
-import org.apache.camel.kafkaconnector.aws.v2.clients.AWSSDKClientUtils;
+import org.apache.camel.test.infra.aws2.clients.AWSSDKClientUtils;
import software.amazon.awssdk.services.kinesis.KinesisClient;
public class TestKinesisConfiguration extends Kinesis2Configuration {
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/s3/source/CamelAWSS3PropertyFactory.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/s3/source/CamelAWSS3PropertyFactory.java
index bb30071..a59876b 100644
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/s3/source/CamelAWSS3PropertyFactory.java
+++
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/s3/source/CamelAWSS3PropertyFactory.java
@@ -21,9 +21,9 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
-import org.apache.camel.kafkaconnector.aws.common.AWSConfigs;
import org.apache.camel.kafkaconnector.common.EndpointUrlBuilder;
import org.apache.camel.kafkaconnector.common.SourceConnectorPropertyFactory;
+import org.apache.camel.test.infra.aws.common.AWSConfigs;
import software.amazon.awssdk.regions.Region;
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/s3/source/CamelSourceAWSS3ITCase.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/s3/source/CamelSourceAWSS3ITCase.java
index aa0a828..d22244a 100644
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/s3/source/CamelSourceAWSS3ITCase.java
+++
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/s3/source/CamelSourceAWSS3ITCase.java
@@ -21,15 +21,15 @@ import java.io.File;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
-import org.apache.camel.kafkaconnector.aws.common.AWSCommon;
-import org.apache.camel.kafkaconnector.aws.common.AWSConfigs;
-import org.apache.camel.kafkaconnector.aws.common.services.AWSService;
import org.apache.camel.kafkaconnector.aws.v2.clients.AWSSDKClientUtils;
-import org.apache.camel.kafkaconnector.aws.v2.services.AWSServiceFactory;
import org.apache.camel.kafkaconnector.common.AbstractKafkaTest;
import org.apache.camel.kafkaconnector.common.ConnectorPropertyFactory;
import org.apache.camel.kafkaconnector.common.clients.kafka.KafkaClient;
import org.apache.camel.kafkaconnector.common.utils.TestUtils;
+import org.apache.camel.test.infra.aws.common.AWSCommon;
+import org.apache.camel.test.infra.aws.common.AWSConfigs;
+import org.apache.camel.test.infra.aws.common.services.AWSService;
+import org.apache.camel.test.infra.aws2.services.AWSServiceFactory;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/s3/source/TestS3Configuration.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/s3/source/TestS3Configuration.java
index 9898f74..ff7c897 100644
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/s3/source/TestS3Configuration.java
+++
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/s3/source/TestS3Configuration.java
@@ -18,7 +18,7 @@
package org.apache.camel.kafkaconnector.aws.v2.s3.source;
import org.apache.camel.component.aws2.s3.AWS2S3Configuration;
-import org.apache.camel.kafkaconnector.aws.v2.clients.AWSSDKClientUtils;
+import org.apache.camel.test.infra.aws2.clients.AWSSDKClientUtils;
import software.amazon.awssdk.services.s3.S3Client;
public class TestS3Configuration extends AWS2S3Configuration {
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSContainer.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSContainer.java
deleted file mode 100644
index dd8e67c..0000000
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSContainer.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.camel.kafkaconnector.aws.v2.services;
-
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.Arrays;
-import java.util.stream.Collectors;
-
-import
org.apache.camel.kafkaconnector.aws.v2.common.TestAWSCredentialsProvider;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.wait.strategy.Wait;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-
-/*
- The reason we are not using LocalStack containers here is because they bundle
AWS SDK v1. They would
- be added to the classpath during the test and, potentially, cause errors or
cause the code to not
- behave as in runtime.
- */
-public class AWSContainer extends GenericContainer<AWSContainer> {
- private static final String LOCALSTACK_CONTAINER =
"localstack/localstack:0.11.3";
-
- private static final Logger LOG =
LoggerFactory.getLogger(AWSLocalContainerService.class);
- private static final int SERVICE_PORT = 4566;
-
- public AWSContainer(Service...services) {
- this(LOCALSTACK_CONTAINER, services);
- }
-
- public AWSContainer(String container, Service...services) {
- super(container);
-
- String serviceList = Arrays.stream(services)
- .map(Service::serviceName)
- .collect(Collectors.joining(","));
-
- LOG.debug("Creating services {}", serviceList);
- withEnv("SERVICE", serviceList);
- withExposedPorts(SERVICE_PORT);
- waitingFor(Wait.forLogMessage(".*Ready\\.\n", 1));
- }
-
- public AwsCredentialsProvider getCredentialsProvider() {
- return TestAWSCredentialsProvider.CONTAINER_LOCAL_DEFAULT_PROVIDER;
- }
-
- protected String getAmazonHost() {
- return getContainerIpAddress() + ":" + getMappedPort(SERVICE_PORT);
- }
-
- public URI getServiceEndpoint() {
- try {
- String address = String.format("http://%s:%d";,
getContainerIpAddress(), getMappedPort(SERVICE_PORT));
- LOG.debug("Running on service endpoint: {}", address);
-
- return new URI(address);
- } catch (URISyntaxException e) {
- throw new RuntimeException(String.format("Unable to determine the
service endpoint: %s", e.getMessage()), e);
- }
-
- }
-}
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSKinesisLocalContainerService.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSKinesisLocalContainerService.java
deleted file mode 100644
index 99b03f7..0000000
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSKinesisLocalContainerService.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.camel.kafkaconnector.aws.v2.services;
-
-import
org.apache.camel.kafkaconnector.aws.v2.common.TestAWSCredentialsProvider;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import software.amazon.awssdk.regions.Region;
-import software.amazon.awssdk.services.kinesis.KinesisClient;
-
-public class AWSKinesisLocalContainerService extends
AWSLocalContainerService<KinesisClient> {
- private static final Logger LOG =
LoggerFactory.getLogger(AWSKinesisLocalContainerService.class);
-
- public AWSKinesisLocalContainerService() {
- super(Service.KINESIS);
-
- LOG.info("Initializing the local AWS services");
- getContainer().start();
- }
-
- @Override
- public KinesisClient getClient() {
- Region region = Region.US_EAST_1;
- KinesisClient client = KinesisClient.builder()
- .region(region)
-
.credentialsProvider(TestAWSCredentialsProvider.CONTAINER_LOCAL_DEFAULT_PROVIDER)
- .endpointOverride(getServiceEndpoint())
- .build();
-
- return client;
- }
-}
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSLocalContainerService.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSLocalContainerService.java
deleted file mode 100644
index bf194fa..0000000
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSLocalContainerService.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.camel.kafkaconnector.aws.v2.services;
-
-import java.net.URI;
-import java.util.Properties;
-
-import org.apache.camel.kafkaconnector.aws.common.AWSConfigs;
-import org.apache.camel.kafkaconnector.aws.common.services.AWSService;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import software.amazon.awssdk.auth.credentials.AwsCredentials;
-import software.amazon.awssdk.regions.Region;
-
-abstract class AWSLocalContainerService<T> implements AWSService<T> {
- private static final Logger LOG =
LoggerFactory.getLogger(AWSLocalContainerService.class);
- private AWSContainer container;
-
- public AWSLocalContainerService(Service...services) {
- container = new AWSContainer(services);
- }
-
- public AWSLocalContainerService(String containerName, Service...services) {
- container = new AWSContainer(containerName, services);
- }
-
- private String getAmazonHost() {
- return container.getAmazonHost();
- }
-
- protected AWSContainer getContainer() {
- return container;
- }
-
- @Override
- public Properties getConnectionProperties() {
- Properties properties = new Properties();
-
- AwsCredentials credentials =
container.getCredentialsProvider().resolveCredentials();
-
- properties.put(AWSConfigs.ACCESS_KEY, credentials.accessKeyId());
-
- properties.put(AWSConfigs.SECRET_KEY, credentials.secretAccessKey());
-
- properties.put(AWSConfigs.REGION, Region.US_EAST_1.toString());
-
- properties.put(AWSConfigs.AMAZON_AWS_HOST, container.getAmazonHost());
-
- /**
- * We need to set this one. For some sets, when they instantiate the
clients within
- * Camel, they need to know what is the Amazon host being used (ie.:
when creating them
- * using the withEndpointConfiguration()). Because this happens within
Camel, there's
- * no way to pass that information easily. Therefore, the host is set
as a property
- * and read by whatever class/method creates the clients to pass to
Camel.
- *
- * Do not unset.
- */
- System.setProperty(AWSConfigs.AMAZON_AWS_HOST, getAmazonHost());
-
- properties.put(AWSConfigs.PROTOCOL, "http");
-
- return properties;
- }
-
- public URI getServiceEndpoint() {
- return container.getServiceEndpoint();
- }
-
- @Override
- public void initialize() {
- LOG.info("AWS service running at address {}", getServiceEndpoint());
- }
-
- @Override
- public void shutdown() {
- LOG.info("Stopping the local AWS services");
- container.stop();
- }
-}
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSRemoteService.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSRemoteService.java
deleted file mode 100644
index 1bdd4c7..0000000
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSRemoteService.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.camel.kafkaconnector.aws.v2.services;
-
-import java.util.Properties;
-import java.util.function.Supplier;
-
-
-import org.apache.camel.kafkaconnector.aws.common.AWSConfigs;
-import org.apache.camel.kafkaconnector.aws.common.services.AWSService;
-import
org.apache.camel.kafkaconnector.aws.v2.common.TestAWSCredentialsProvider;
-import software.amazon.awssdk.auth.credentials.AwsCredentials;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.regions.Region;
-
-public class AWSRemoteService<T> implements AWSService<T> {
- private static final AwsCredentialsProvider CREDENTIALS_PROVIDER = new
TestAWSCredentialsProvider();
- private Supplier<T> remoteClientSupplier;
-
- public AWSRemoteService(Supplier<T> remoteClientSupplier) {
- this.remoteClientSupplier = remoteClientSupplier;
- }
-
-
- @Override
- public T getClient() {
- return remoteClientSupplier.get();
- }
-
- @Override
- public Properties getConnectionProperties() {
- Properties properties = new Properties();
-
- AwsCredentials credentials = CREDENTIALS_PROVIDER.resolveCredentials();
-
- properties.put(AWSConfigs.ACCESS_KEY, credentials.accessKeyId());
- properties.put(AWSConfigs.SECRET_KEY, credentials.secretAccessKey());
- properties.put(AWSConfigs.REGION, Region.US_EAST_1.toString());
-
- return properties;
- }
-
- @Override
- public void initialize() {
-
- }
-
- @Override
- public void shutdown() {
-
- }
-}
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSS3LocalContainerService.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSS3LocalContainerService.java
deleted file mode 100644
index c9bef95..0000000
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSS3LocalContainerService.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.camel.kafkaconnector.aws.v2.services;
-
-import
org.apache.camel.kafkaconnector.aws.v2.common.TestAWSCredentialsProvider;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import software.amazon.awssdk.regions.Region;
-import software.amazon.awssdk.services.s3.S3Client;
-
-public class AWSS3LocalContainerService extends
AWSLocalContainerService<S3Client> {
- private static final Logger LOG =
LoggerFactory.getLogger(AWSS3LocalContainerService.class);
-
- public AWSS3LocalContainerService() {
- super(Service.S3);
-
- LOG.info("Initializing the local AWS services");
- getContainer().start();
- }
-
-
- @Override
- public S3Client getClient() {
- Region region = Region.US_EAST_1;
-
- return S3Client.builder()
- .region(region)
-
.credentialsProvider(TestAWSCredentialsProvider.CONTAINER_LOCAL_DEFAULT_PROVIDER)
- .endpointOverride(getServiceEndpoint())
- .build();
- }
-}
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSSQSLocalContainerService.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSSQSLocalContainerService.java
deleted file mode 100644
index 9c76a9a..0000000
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSSQSLocalContainerService.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.camel.kafkaconnector.aws.v2.services;
-
-import
org.apache.camel.kafkaconnector.aws.v2.common.TestAWSCredentialsProvider;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import software.amazon.awssdk.regions.Region;
-import software.amazon.awssdk.services.sqs.SqsClient;
-
-public class AWSSQSLocalContainerService extends
AWSLocalContainerService<SqsClient> {
- private static final Logger LOG =
LoggerFactory.getLogger(AWSSQSLocalContainerService.class);
-
- public AWSSQSLocalContainerService() {
- // Current latest container - localstack/localstack:0.11.3 - is broken
for SQS
- // therefore uses an older version
- super("localstack/localstack:0.11.2", Service.SQS);
-
- LOG.info("Initializing the local AWS services");
- getContainer().start();
- }
-
- @Override
- public SqsClient getClient() {
- Region region = Region.US_EAST_1;
-
- return SqsClient.builder()
- .region(region)
-
.credentialsProvider(TestAWSCredentialsProvider.CONTAINER_LOCAL_DEFAULT_PROVIDER)
- .endpointOverride(getServiceEndpoint())
- .build();
- }
-}
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSServiceFactory.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSServiceFactory.java
deleted file mode 100644
index 7429b21..0000000
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/AWSServiceFactory.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.camel.kafkaconnector.aws.v2.services;
-
-import org.apache.camel.kafkaconnector.aws.common.services.AWSService;
-import org.apache.camel.kafkaconnector.aws.v2.clients.AWSSDKClientUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import software.amazon.awssdk.core.SdkSystemSetting;
-import software.amazon.awssdk.services.kinesis.KinesisClient;
-import software.amazon.awssdk.services.s3.S3Client;
-import software.amazon.awssdk.services.sqs.SqsClient;
-
-public final class AWSServiceFactory {
- private static final Logger LOG =
LoggerFactory.getLogger(AWSServiceFactory.class);
-
- private AWSServiceFactory() {
-
- }
-
- private static String getInstanceTypeName(String awsInstanceType) {
- return awsInstanceType == null ? "default" : awsInstanceType;
- }
-
- public static AWSService<KinesisClient> createKinesisService() {
- String awsInstanceType =
System.getProperty("aws-service.kinesis.instance.type");
- LOG.info("Creating a {} AWS kinesis instance",
getInstanceTypeName(awsInstanceType));
-
-
- if (awsInstanceType == null ||
awsInstanceType.equals("local-aws-container")) {
-
- System.setProperty(SdkSystemSetting.CBOR_ENABLED.property(),
"false");
-
- return new AWSKinesisLocalContainerService();
- }
-
- if (awsInstanceType.equals("remote")) {
- return new AWSRemoteService<>(AWSSDKClientUtils::newKinesisClient);
- }
-
- LOG.error("Invalid AWS instance type: {}. Must be either 'remote' or
'local-aws-container'",
- awsInstanceType);
- throw new UnsupportedOperationException("Invalid AWS instance type");
- }
-
- public static AWSService<SqsClient> createSQSService() {
- String awsInstanceType =
System.getProperty("aws-service.instance.type");
- LOG.info("Creating a {} AWS SQS instance",
getInstanceTypeName(awsInstanceType));
-
-
- if (awsInstanceType == null ||
awsInstanceType.equals("local-aws-container")) {
- return new AWSSQSLocalContainerService();
- }
-
- if (awsInstanceType.equals("remote")) {
- return new AWSRemoteService<>(AWSSDKClientUtils::newSQSClient);
- }
-
- LOG.error("Invalid AWS instance type: {}. Must be either 'remote' or
'local-aws-container'",
- awsInstanceType);
- throw new UnsupportedOperationException("Invalid AWS instance type");
- }
-
- public static AWSService<S3Client> createS3Service() {
- String awsInstanceType =
System.getProperty("aws-service.instance.type");
- LOG.info("Creating a {} AWS S3 instance", awsInstanceType);
-
- if (awsInstanceType == null ||
awsInstanceType.equals("local-aws-container")) {
- return new AWSS3LocalContainerService();
- }
-
- if (awsInstanceType.equals("remote")) {
- return new AWSRemoteService<>(AWSSDKClientUtils::newS3Client);
- }
-
- LOG.error("Invalid AWS instance type: {}. Must be either 'remote' or
'local-aws-container'",
- awsInstanceType);
- throw new UnsupportedOperationException("Invalid AWS instance type");
- }
-}
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/Service.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/Service.java
deleted file mode 100644
index 8c6a3b3..0000000
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/services/Service.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.camel.kafkaconnector.aws.v2.services;
-
-public enum Service {
- KINESIS("kinesis"),
- SQS("sqs"),
- S3("s3");
-
- private final String serviceName;
-
- Service(String serviceName) {
- this.serviceName = serviceName;
- }
-
- public String getServiceName() {
- return serviceName;
- }
-
- public static String serviceName(Service service) {
- return service.serviceName;
- }
-}
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/sink/CamelAWSSQSPropertyFactory.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/sink/CamelAWSSQSPropertyFactory.java
index a49e347..0ac2d5c 100644
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/sink/CamelAWSSQSPropertyFactory.java
+++
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/sink/CamelAWSSQSPropertyFactory.java
@@ -21,9 +21,9 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
-import org.apache.camel.kafkaconnector.aws.common.AWSConfigs;
import org.apache.camel.kafkaconnector.common.EndpointUrlBuilder;
import org.apache.camel.kafkaconnector.common.SinkConnectorPropertyFactory;
+import org.apache.camel.test.infra.aws.common.AWSConfigs;
import software.amazon.awssdk.regions.Region;
/**
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/sink/CamelSinkAWSSQSITCase.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/sink/CamelSinkAWSSQSITCase.java
index d913696..23a1551 100644
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/sink/CamelSinkAWSSQSITCase.java
+++
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/sink/CamelSinkAWSSQSITCase.java
@@ -24,15 +24,15 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
-import org.apache.camel.kafkaconnector.aws.common.AWSCommon;
-import org.apache.camel.kafkaconnector.aws.common.AWSConfigs;
-import org.apache.camel.kafkaconnector.aws.common.services.AWSService;
import org.apache.camel.kafkaconnector.aws.v2.clients.AWSSQSClient;
-import org.apache.camel.kafkaconnector.aws.v2.services.AWSServiceFactory;
import org.apache.camel.kafkaconnector.common.AbstractKafkaTest;
import org.apache.camel.kafkaconnector.common.ConnectorPropertyFactory;
import org.apache.camel.kafkaconnector.common.clients.kafka.KafkaClient;
import org.apache.camel.kafkaconnector.common.utils.TestUtils;
+import org.apache.camel.test.infra.aws.common.AWSCommon;
+import org.apache.camel.test.infra.aws.common.AWSConfigs;
+import org.apache.camel.test.infra.aws.common.services.AWSService;
+import org.apache.camel.test.infra.aws2.services.AWSServiceFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/source/CamelAWSSQSPropertyFactory.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/source/CamelAWSSQSPropertyFactory.java
index 53737a7..6e8d946 100644
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/source/CamelAWSSQSPropertyFactory.java
+++
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/source/CamelAWSSQSPropertyFactory.java
@@ -21,9 +21,9 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
-import org.apache.camel.kafkaconnector.aws.common.AWSConfigs;
import org.apache.camel.kafkaconnector.common.EndpointUrlBuilder;
import org.apache.camel.kafkaconnector.common.SourceConnectorPropertyFactory;
+import org.apache.camel.test.infra.aws.common.AWSConfigs;
import software.amazon.awssdk.regions.Region;
/**
diff --git
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/source/CamelSourceAWSSQSITCase.java
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/source/CamelSourceAWSSQSITCase.java
index dcb1b88..de1665b 100644
---
a/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/source/CamelSourceAWSSQSITCase.java
+++
b/tests/itests-aws-v2/src/test/java/org/apache/camel/kafkaconnector/aws/v2/sqs/source/CamelSourceAWSSQSITCase.java
@@ -20,15 +20,15 @@ package org.apache.camel.kafkaconnector.aws.v2.sqs.source;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
-import org.apache.camel.kafkaconnector.aws.common.AWSCommon;
-import org.apache.camel.kafkaconnector.aws.common.AWSConfigs;
-import org.apache.camel.kafkaconnector.aws.common.services.AWSService;
import org.apache.camel.kafkaconnector.aws.v2.clients.AWSSQSClient;
-import org.apache.camel.kafkaconnector.aws.v2.services.AWSServiceFactory;
import org.apache.camel.kafkaconnector.common.AbstractKafkaTest;
import org.apache.camel.kafkaconnector.common.ConnectorPropertyFactory;
import org.apache.camel.kafkaconnector.common.clients.kafka.KafkaClient;
import org.apache.camel.kafkaconnector.common.utils.TestUtils;
+import org.apache.camel.test.infra.aws.common.AWSCommon;
+import org.apache.camel.test.infra.aws.common.AWSConfigs;
+import org.apache.camel.test.infra.aws.common.services.AWSService;
+import org.apache.camel.test.infra.aws2.services.AWSServiceFactory;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
