You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by or...@apache.org on 2021/02/04 14:12:47 UTC
[camel-kafka-connector] 04/22: Convert the Azure storage queue
tests to the new reusable sink test base class
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch camel-master
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
commit 995b8078ceb673e99c91e91c26a96622a8a23187
Author: Otavio Rodolfo Piske <op...@redhat.com>
AuthorDate: Wed Feb 3 11:27:01 2021 +0100
Convert the Azure storage queue tests to the new reusable sink test base class
---
.../sink/CamelSinkAzureStorageQueueITCase.java | 82 ++++++++++------------
1 file changed, 38 insertions(+), 44 deletions(-)
diff --git a/tests/itests-azure-storage-queue/src/test/java/org/apache/camel/kafkaconnector/azure/storage/queue/sink/CamelSinkAzureStorageQueueITCase.java b/tests/itests-azure-storage-queue/src/test/java/org/apache/camel/kafkaconnector/azure/storage/queue/sink/CamelSinkAzureStorageQueueITCase.java
index a7c16ed..d447703 100644
--- a/tests/itests-azure-storage-queue/src/test/java/org/apache/camel/kafkaconnector/azure/storage/queue/sink/CamelSinkAzureStorageQueueITCase.java
+++ b/tests/itests-azure-storage-queue/src/test/java/org/apache/camel/kafkaconnector/azure/storage/queue/sink/CamelSinkAzureStorageQueueITCase.java
@@ -17,20 +17,17 @@
package org.apache.camel.kafkaconnector.azure.storage.queue.sink;
-import java.io.IOException;
-import java.util.HashMap;
import java.util.Map;
-import java.util.concurrent.ExecutionException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import com.azure.storage.queue.QueueClient;
import com.azure.storage.queue.QueueServiceClient;
import com.azure.storage.queue.models.PeekedMessageItem;
-import org.apache.camel.kafkaconnector.CamelSinkTask;
import org.apache.camel.kafkaconnector.azure.storage.queue.common.TestQueueConfiguration;
import org.apache.camel.kafkaconnector.azure.storage.services.AzureStorageClientUtils;
-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.test.CamelSinkTestSupport;
import org.apache.camel.kafkaconnector.common.utils.TestUtils;
import org.apache.camel.test.infra.azure.common.AzureCredentialsHolder;
import org.apache.camel.test.infra.azure.common.services.AzureService;
@@ -45,9 +42,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.fail;
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
-public class CamelSinkAzureStorageQueueITCase extends AbstractKafkaTest {
+public class CamelSinkAzureStorageQueueITCase extends CamelSinkTestSupport {
@RegisterExtension
public static AzureService service = AzureStorageQueueServiceFactory.createAzureService();
@@ -56,6 +54,7 @@ public class CamelSinkAzureStorageQueueITCase extends AbstractKafkaTest {
private QueueServiceClient client;
private QueueClient queueClient;
private String queueName;
+ private String topicName;
private int expect = 10;
private int received;
@@ -66,6 +65,8 @@ public class CamelSinkAzureStorageQueueITCase extends AbstractKafkaTest {
@BeforeEach
public void setUp() {
+ topicName = getTopicForTest(this);
+
client = AzureStorageClientUtils.getClient();
queueName = "test-queue" + TestUtils.randomWithRange(0, 100);
@@ -80,6 +81,30 @@ public class CamelSinkAzureStorageQueueITCase extends AbstractKafkaTest {
}
}
+ @Override
+ protected Map<String, String> messageHeaders(String text, int current) {
+ return null;
+ }
+
+ @Override
+ protected void consumeMessages(CountDownLatch latch) {
+ try {
+ consume();
+ } finally {
+ latch.countDown();
+ }
+ }
+
+ @Override
+ protected void verifyMessages(CountDownLatch latch) throws InterruptedException {
+ if (latch.await(120, TimeUnit.SECONDS)) {
+ assertEquals(expect, received,
+ "Didn't process the expected amount of messages: " + received + " != " + expect);
+ } else {
+ fail("Failed to receive the messages within the specified time");
+ }
+ }
+
private void acknowledgeReceived(PeekedMessageItem peekedMessageItem) {
received++;
LOG.info("Received: {}", peekedMessageItem.getMessageText());
@@ -97,72 +122,41 @@ public class CamelSinkAzureStorageQueueITCase extends AbstractKafkaTest {
int count = queueClient.getProperties().getApproximateMessagesCount();
queueClient.peekMessages(count, null, null).forEach(this::acknowledgeReceived);
-
- }
-
- private void putRecords() {
- Map<String, String> messageParameters = new HashMap<>();
- KafkaClient<String, String> kafkaClient = new KafkaClient<>(getKafkaService().getBootstrapServers());
-
- for (int i = 0; i < expect; i++) {
- try {
- // This is for 3.4 only. From 3.5 and newer, the text is taken from the body
- messageParameters.put(CamelSinkTask.HEADER_CAMEL_PREFIX + "CamelAzureStorageQueueMessageText", "test " + i);
-
- kafkaClient.produce(TestUtils.getDefaultTestTopic(this.getClass()), "test " + i, messageParameters);
- } catch (ExecutionException e) {
- LOG.error("Unable to produce messages: {}", e.getMessage(), e);
- } catch (InterruptedException e) {
- break;
- }
- }
- }
-
-
- public void runTest(ConnectorPropertyFactory connectorPropertyFactory) throws ExecutionException, InterruptedException {
- connectorPropertyFactory.log();
- getKafkaConnectService().initializeConnectorBlocking(connectorPropertyFactory, 1);
-
- putRecords();
-
- consume();
-
- assertEquals(expect, received, "Did not receive the same amount of messages that were sent");
}
@Test
@Timeout(90)
- public void testBasicSendReceive() throws InterruptedException, ExecutionException, IOException {
+ public void testBasicSendReceive() throws Exception {
AzureCredentialsHolder azureCredentialsHolder = service.azureCredentials();
ConnectorPropertyFactory connectorPropertyFactory = CamelSinkAzureStorageQueuePropertyFactory
.basic()
.withConfiguration(TestQueueConfiguration.class.getName())
- .withTopics(TestUtils.getDefaultTestTopic(this.getClass()))
+ .withTopics(topicName)
.withAccessKey(azureCredentialsHolder.accountKey())
.withAccountName(azureCredentialsHolder.accountName())
.withOperation("sendMessage")
.withQueueName(queueName);
- runTest(connectorPropertyFactory);
+ runTest(connectorPropertyFactory, topicName, expect);
}
@Test
@Timeout(90)
- public void testBasicSendReceiveUrl() throws InterruptedException, ExecutionException, IOException {
+ public void testBasicSendReceiveUrl() throws Exception {
AzureCredentialsHolder azureCredentialsHolder = service.azureCredentials();
ConnectorPropertyFactory connectorPropertyFactory = CamelSinkAzureStorageQueuePropertyFactory
.basic()
- .withTopics(TestUtils.getDefaultTestTopic(this.getClass()))
+ .withTopics(topicName)
.withConfiguration(TestQueueConfiguration.class.getName())
.withUrl(azureCredentialsHolder.accountName() + "/" + queueName)
.append("accessKey", azureCredentialsHolder.accountKey())
.append("operation", "sendMessage")
.buildUrl();
- runTest(connectorPropertyFactory);
+ runTest(connectorPropertyFactory, topicName, expect);
}
}