You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2021/01/29 09:16:44 UTC
[camel-kafka-connector] branch master updated: Added Azure Storage
Queue source test case
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 c116cce Added Azure Storage Queue source test case
c116cce is described below
commit c116ccefea512c81301524dfa66516074c3ced16
Author: Otavio Rodolfo Piske <op...@redhat.com>
AuthorDate: Fri Jan 29 09:43:44 2021 +0100
Added Azure Storage Queue source test case
---
.../{sink => common}/TestQueueConfiguration.java | 2 +-
.../sink/CamelSinkAzureStorageQueueITCase.java | 1 +
.../CamelSourceAzureStorageQueueITCase.java} | 96 +++++++---------------
...amelSourceAzureStorageQueuePropertyFactory.java | 52 ++++++++++++
4 files changed, 83 insertions(+), 68 deletions(-)
diff --git a/tests/itests-azure-storage-queue/src/test/java/org/apache/camel/kafkaconnector/azure/storage/queue/sink/TestQueueConfiguration.java b/tests/itests-azure-storage-queue/src/test/java/org/apache/camel/kafkaconnector/azure/storage/queue/common/TestQueueConfiguration.java
similarity index 95%
rename from tests/itests-azure-storage-queue/src/test/java/org/apache/camel/kafkaconnector/azure/storage/queue/sink/TestQueueConfiguration.java
rename to tests/itests-azure-storage-queue/src/test/java/org/apache/camel/kafkaconnector/azure/storage/queue/common/TestQueueConfiguration.java
index d91c43e..570c633 100644
--- a/tests/itests-azure-storage-queue/src/test/java/org/apache/camel/kafkaconnector/azure/storage/queue/sink/TestQueueConfiguration.java
+++ b/tests/itests-azure-storage-queue/src/test/java/org/apache/camel/kafkaconnector/azure/storage/queue/common/TestQueueConfiguration.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.camel.kafkaconnector.azure.storage.queue.sink;
+package org.apache.camel.kafkaconnector.azure.storage.queue.common;
import com.azure.storage.queue.QueueServiceClient;
import org.apache.camel.component.azure.storage.queue.QueueConfiguration;
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 b0636ff..a7c16ed 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
@@ -26,6 +26,7 @@ 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;
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/source/CamelSourceAzureStorageQueueITCase.java
similarity index 56%
copy from tests/itests-azure-storage-queue/src/test/java/org/apache/camel/kafkaconnector/azure/storage/queue/sink/CamelSinkAzureStorageQueueITCase.java
copy to tests/itests-azure-storage-queue/src/test/java/org/apache/camel/kafkaconnector/azure/storage/queue/source/CamelSourceAzureStorageQueueITCase.java
index b0636ff..b862255 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/source/CamelSourceAzureStorageQueueITCase.java
@@ -15,17 +15,14 @@
* limitations under the License.
*/
-package org.apache.camel.kafkaconnector.azure.storage.queue.sink;
+package org.apache.camel.kafkaconnector.azure.storage.queue.source;
import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
import java.util.concurrent.ExecutionException;
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;
@@ -34,6 +31,7 @@ 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;
import org.apache.camel.test.infra.azure.storage.queue.services.AzureStorageQueueServiceFactory;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -46,11 +44,10 @@ import org.slf4j.LoggerFactory;
import static org.junit.jupiter.api.Assertions.assertEquals;
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
-public class CamelSinkAzureStorageQueueITCase extends AbstractKafkaTest {
+public class CamelSourceAzureStorageQueueITCase extends AbstractKafkaTest {
@RegisterExtension
public static AzureService service = AzureStorageQueueServiceFactory.createAzureService();
-
- private static final Logger LOG = LoggerFactory.getLogger(CamelSinkAzureStorageQueueITCase.class);
+ private static final Logger LOG = LoggerFactory.getLogger(CamelSourceAzureStorageQueueITCase.class);
private QueueServiceClient client;
private QueueClient queueClient;
@@ -79,54 +76,37 @@ public class CamelSinkAzureStorageQueueITCase extends AbstractKafkaTest {
}
}
- private void acknowledgeReceived(PeekedMessageItem peekedMessageItem) {
- received++;
- LOG.info("Received: {}", peekedMessageItem.getMessageText());
- }
-
- private boolean canConsume() {
- return queueClient.getProperties().getApproximateMessagesCount() >= expect;
- }
-
- private void consume() {
- LOG.debug("Created the consumer ...");
- TestUtils.waitFor(this::canConsume);
-
- LOG.debug("About to receive messages");
- int count = queueClient.getProperties().getApproximateMessagesCount();
-
- queueClient.peekMessages(count, null, null).forEach(this::acknowledgeReceived);
-
+ private void sendMessages() {
+ for (int i = 0; i < expect; i++) {
+ queueClient.sendMessage("Test message " + i);
+ }
}
- private void putRecords() {
- Map<String, String> messageParameters = new HashMap<>();
- KafkaClient<String, String> kafkaClient = new KafkaClient<>(getKafkaService().getBootstrapServers());
+ private boolean checkRecord(ConsumerRecord<String, String> record) {
+ LOG.debug("Received: {}", record.value());
+ received++;
- 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;
- }
+ if (received == expect) {
+ return false;
}
- }
+ return true;
+ }
- public void runTest(ConnectorPropertyFactory connectorPropertyFactory) throws ExecutionException, InterruptedException {
+ public void runtTest(ConnectorPropertyFactory connectorPropertyFactory) throws ExecutionException, InterruptedException {
connectorPropertyFactory.log();
- getKafkaConnectService().initializeConnectorBlocking(connectorPropertyFactory, 1);
+ getKafkaConnectService().initializeConnector(connectorPropertyFactory);
- putRecords();
+ sendMessages();
- consume();
+ LOG.debug("Initialized the connector and put the data for the test execution");
- assertEquals(expect, received, "Did not receive the same amount of messages that were sent");
+ LOG.debug("Creating the consumer ...");
+ KafkaClient<String, String> kafkaClient = new KafkaClient<>(getKafkaService().getBootstrapServers());
+ kafkaClient.consume(TestUtils.getDefaultTestTopic(this.getClass()), this::checkRecord);
+ LOG.debug("Created the consumer ...");
+
+ assertEquals(received, expect, "Didn't process the expected amount of messages");
}
@@ -135,33 +115,15 @@ public class CamelSinkAzureStorageQueueITCase extends AbstractKafkaTest {
public void testBasicSendReceive() throws InterruptedException, ExecutionException, IOException {
AzureCredentialsHolder azureCredentialsHolder = service.azureCredentials();
- ConnectorPropertyFactory connectorPropertyFactory = CamelSinkAzureStorageQueuePropertyFactory
+ ConnectorPropertyFactory connectorPropertyFactory = CamelSourceAzureStorageQueuePropertyFactory
.basic()
.withConfiguration(TestQueueConfiguration.class.getName())
- .withTopics(TestUtils.getDefaultTestTopic(this.getClass()))
+ .withKafkaTopic(TestUtils.getDefaultTestTopic(this.getClass()))
.withAccessKey(azureCredentialsHolder.accountKey())
.withAccountName(azureCredentialsHolder.accountName())
- .withOperation("sendMessage")
.withQueueName(queueName);
- runTest(connectorPropertyFactory);
+ runtTest(connectorPropertyFactory);
}
-
- @Test
- @Timeout(90)
- public void testBasicSendReceiveUrl() throws InterruptedException, ExecutionException, IOException {
- AzureCredentialsHolder azureCredentialsHolder = service.azureCredentials();
-
- ConnectorPropertyFactory connectorPropertyFactory = CamelSinkAzureStorageQueuePropertyFactory
- .basic()
- .withTopics(TestUtils.getDefaultTestTopic(this.getClass()))
- .withConfiguration(TestQueueConfiguration.class.getName())
- .withUrl(azureCredentialsHolder.accountName() + "/" + queueName)
- .append("accessKey", azureCredentialsHolder.accountKey())
- .append("operation", "sendMessage")
- .buildUrl();
-
- runTest(connectorPropertyFactory);
- }
}
diff --git a/tests/itests-azure-storage-queue/src/test/java/org/apache/camel/kafkaconnector/azure/storage/queue/source/CamelSourceAzureStorageQueuePropertyFactory.java b/tests/itests-azure-storage-queue/src/test/java/org/apache/camel/kafkaconnector/azure/storage/queue/source/CamelSourceAzureStorageQueuePropertyFactory.java
new file mode 100644
index 0000000..13024cf
--- /dev/null
+++ b/tests/itests-azure-storage-queue/src/test/java/org/apache/camel/kafkaconnector/azure/storage/queue/source/CamelSourceAzureStorageQueuePropertyFactory.java
@@ -0,0 +1,52 @@
+/*
+ * 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.azure.storage.queue.source;
+
+import org.apache.camel.kafkaconnector.common.SourceConnectorPropertyFactory;
+
+public class CamelSourceAzureStorageQueuePropertyFactory extends SourceConnectorPropertyFactory<CamelSourceAzureStorageQueuePropertyFactory> {
+ public CamelSourceAzureStorageQueuePropertyFactory withAccountName(String value) {
+ return setProperty("camel.source.path.accountName", value);
+ }
+
+ public CamelSourceAzureStorageQueuePropertyFactory withQueueName(String value) {
+ return setProperty("camel.source.path.queueName", value);
+ }
+
+ public CamelSourceAzureStorageQueuePropertyFactory withAccessKey(String value) {
+ return setProperty("camel.component.azure-storage-queue.accessKey", value);
+ }
+
+ public CamelSourceAzureStorageQueuePropertyFactory withConfiguration(String configurationClass) {
+ return setProperty("camel.component.azure-storage-queue.configuration", classRef(configurationClass));
+ }
+
+ public CamelSourceAzureStorageQueuePropertyFactory withOperation(String value) {
+ return setProperty("camel.component.azure-storage-queue.operation", value);
+ }
+
+ public static CamelSourceAzureStorageQueuePropertyFactory basic() {
+ return new CamelSourceAzureStorageQueuePropertyFactory()
+ .withTasksMax(1)
+ .withName("CamelAzurequeueSourceConnector")
+ .withConnectorClass("org.apache.camel.kafkaconnector.azurestoragequeue.CamelAzurestoragequeueSourceConnector")
+ .withKeyConverterClass("org.apache.kafka.connect.storage.StringConverter")
+ .withValueConverterClass("org.apache.kafka.connect.storage.StringConverter");
+
+ }
+}