This is an automated email from the ASF dual-hosted git repository. lhotari pushed a commit to branch branch-3.3 in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 80090b259a48435203bbd92fd27811bdcb4618e0 Author: Lari Hotari <[email protected]> AuthorDate: Wed Aug 7 13:46:38 2024 +0300 [fix][build] Remove unnecessary Oracle maven repository from pom.xml (#23132) (cherry picked from commit 175ea005747ea7e229b2f492b039cf51007421ad) --- pom.xml | 8 --- pulsar-io/rabbitmq/pom.xml | 14 +++++ .../pulsar/io/rabbitmq/RabbitMQBrokerManager.java | 18 +------ .../pulsar/io/rabbitmq/sink/RabbitMQSinkTest.java | 9 +++- .../io/rabbitmq/source/RabbitMQSourceTest.java | 10 +++- pulsar-io/rabbitmq/src/test/resources/qpid.json | 59 +++++++++++++++++----- 6 files changed, 75 insertions(+), 43 deletions(-) diff --git a/pom.xml b/pom.xml index 768e82863d9..6f3a4c44ed8 100644 --- a/pom.xml +++ b/pom.xml @@ -2794,13 +2794,5 @@ flexible messaging model and an intuitive client API.</description> <enabled>false</enabled> </snapshots> </repository> - <!-- For the BDB JE dependency --> - <repository> - <id>oracle.releases</id> - <url>https://download.oracle.com/maven</url> - <snapshots> - <enabled>false</enabled> - </snapshots> - </repository> </repositories> </project> diff --git a/pulsar-io/rabbitmq/pom.xml b/pulsar-io/rabbitmq/pom.xml index 8b0aab5f747..fff4cdb4194 100644 --- a/pulsar-io/rabbitmq/pom.xml +++ b/pulsar-io/rabbitmq/pom.xml @@ -85,6 +85,20 @@ <artifactId>qpid-broker</artifactId> <version>9.2.0</version> <scope>test</scope> + <exclusions> + <exclusion> + <groupId>org.apache.qpid</groupId> + <artifactId>qpid-bdbstore</artifactId> + </exclusion> + <exclusion> + <groupId>org.apache.qpid</groupId> + <artifactId>qpid-broker-plugins-amqp-1-0-protocol-bdb-link-store</artifactId> + </exclusion> + <exclusion> + <groupId>org.apache.qpid</groupId> + <artifactId>qpid-broker-plugins-derby-store</artifactId> + </exclusion> + </exclusions> </dependency> <dependency> <groupId>org.awaitility</groupId> diff --git a/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/RabbitMQBrokerManager.java b/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/RabbitMQBrokerManager.java index 83331bf7de8..4ff8c61e4f4 100644 --- a/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/RabbitMQBrokerManager.java +++ b/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/RabbitMQBrokerManager.java @@ -18,8 +18,6 @@ */ package org.apache.pulsar.io.rabbitmq; -import java.io.File; -import java.io.FileOutputStream; import java.nio.file.Files; import java.nio.file.Path; import java.util.HashMap; @@ -42,28 +40,14 @@ public class RabbitMQBrokerManager { Map<String, Object> getBrokerOptions(String port) throws Exception { Path tmpFolder = Files.createTempDirectory("qpidWork"); - Path homeFolder = Files.createTempDirectory("qpidHome"); - File etc = new File(homeFolder.toFile(), "etc"); - etc.mkdir(); - FileOutputStream fos = new FileOutputStream(new File(etc, "passwd")); - fos.write("guest:guest\n".getBytes()); - fos.close(); - Map<String, Object> config = new HashMap<>(); config.put("qpid.work_dir", tmpFolder.toAbsolutePath().toString()); config.put("qpid.amqp_port", port); - config.put("qpid.home_dir", homeFolder.toAbsolutePath().toString()); - String configPath = getFile("qpid.json").getAbsolutePath(); Map<String, Object> context = new HashMap<>(); - context.put(SystemConfig.INITIAL_CONFIGURATION_LOCATION, configPath); + context.put(SystemConfig.INITIAL_CONFIGURATION_LOCATION, "classpath:qpid.json"); context.put(SystemConfig.TYPE, "Memory"); context.put(SystemConfig.CONTEXT, config); return context; } - - private File getFile(String name) { - ClassLoader classLoader = getClass().getClassLoader(); - return new File(classLoader.getResource(name).getFile()); - } } diff --git a/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/sink/RabbitMQSinkTest.java b/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/sink/RabbitMQSinkTest.java index 3b20c61f826..f03a36ce114 100644 --- a/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/sink/RabbitMQSinkTest.java +++ b/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/sink/RabbitMQSinkTest.java @@ -18,12 +18,15 @@ */ package org.apache.pulsar.io.rabbitmq.sink; +import static org.mockito.Mockito.mock; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.HashMap; import java.util.Map; import java.util.Optional; import org.apache.pulsar.functions.api.Record; import org.apache.pulsar.functions.instance.SinkRecord; +import org.apache.pulsar.io.core.SinkContext; import org.apache.pulsar.io.rabbitmq.RabbitMQBrokerManager; import org.apache.pulsar.io.rabbitmq.RabbitMQSink; import org.awaitility.Awaitility; @@ -46,7 +49,7 @@ public class RabbitMQSinkTest { } @Test - public void TestOpenAndWriteSink() throws Exception { + public void testOpenAndWriteSink() throws Exception { Map<String, Object> configs = new HashMap<>(); configs.put("host", "localhost"); configs.put("port", "5673"); @@ -66,7 +69,9 @@ public class RabbitMQSinkTest { // open should success // rabbitmq service may need time to initialize - Awaitility.await().ignoreExceptions().untilAsserted(() -> sink.open(configs, null)); + SinkContext sinkContext = mock(SinkContext.class); + Awaitility.await().ignoreExceptions().pollDelay(Duration.ofSeconds(1)) + .untilAsserted(() -> sink.open(configs, sinkContext)); // write should success Record<byte[]> record = build("test-topic", "fakeKey", "fakeValue", "fakeRoutingKey"); diff --git a/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/source/RabbitMQSourceTest.java b/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/source/RabbitMQSourceTest.java index 2771185b841..08869e018c6 100644 --- a/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/source/RabbitMQSourceTest.java +++ b/pulsar-io/rabbitmq/src/test/java/org/apache/pulsar/io/rabbitmq/source/RabbitMQSourceTest.java @@ -18,6 +18,9 @@ */ package org.apache.pulsar.io.rabbitmq.source; +import static org.mockito.Mockito.mock; +import java.time.Duration; +import org.apache.pulsar.io.core.SourceContext; import org.apache.pulsar.io.rabbitmq.RabbitMQBrokerManager; import org.apache.pulsar.io.rabbitmq.RabbitMQSource; import org.awaitility.Awaitility; @@ -44,7 +47,7 @@ public class RabbitMQSourceTest { } @Test - public void TestOpenAndWriteSink() throws Exception { + public void testOpenAndWriteSink() throws Exception { Map<String, Object> configs = new HashMap<>(); configs.put("host", "localhost"); configs.put("port", "5672"); @@ -66,8 +69,11 @@ public class RabbitMQSourceTest { // open should success // rabbitmq service may need time to initialize - Awaitility.await().ignoreExceptions().untilAsserted(() -> source.open(configs, null)); + SourceContext sourceContext = mock(SourceContext.class); + Awaitility.await().ignoreExceptions().pollDelay(Duration.ofSeconds(1)) + .untilAsserted(() -> source.open(configs, sourceContext)); source.close(); } + } diff --git a/pulsar-io/rabbitmq/src/test/resources/qpid.json b/pulsar-io/rabbitmq/src/test/resources/qpid.json index 6a0381f6ddc..419e9cc1e4a 100644 --- a/pulsar-io/rabbitmq/src/test/resources/qpid.json +++ b/pulsar-io/rabbitmq/src/test/resources/qpid.json @@ -1,25 +1,57 @@ { - "name": "EmbeddedBroker", + "name": "${broker.name}", "modelVersion": "2.0", - "storeVersion": 1, "authenticationproviders": [ { - "name": "noPassword", - "type": "Anonymous", - "secureOnlyMechanisms": [] - }, + "name": "plain", + "type": "Plain", + "secureOnlyMechanisms": [], + "users": [ + { + "name": "guest", + "password": "guest", + "type": "managed" + } + ] + } + ], + "brokerloggers": [ { - "name": "passwordFile", - "type": "PlainPasswordFile", - "path": "${qpid.home_dir}${file.separator}etc${file.separator}passwd", - "secureOnlyMechanisms": [] + "name": "console", + "type": "Console", + "brokerloginclusionrules": [ + { + "name": "Root", + "type": "NameAndLevel", + "level": "WARN", + "loggerName": "ROOT" + }, + { + "name": "Qpid", + "type": "NameAndLevel", + "level": "INFO", + "loggerName": "org.apache.qpid.*" + }, + { + "name": "Operational", + "type": "NameAndLevel", + "level": "INFO", + "loggerName": "qpid.message.*" + }, + { + "name": "Statistics", + "type": "NameAndLevel", + "level": "INFO", + "loggerName": "qpid.statistics.*" + } + ] } ], "ports": [ { "name": "AMQP", "port": "${qpid.amqp_port}", - "authenticationProvider": "passwordFile", + "authenticationProvider": "plain", "protocols": [ "AMQP_0_9_1" ] @@ -28,10 +60,9 @@ "virtualhostnodes": [ { "name": "default", - "type": "JSON", + "type": "Memory", "defaultVirtualHostNode": "true", - "virtualHostInitialConfiguration": "${qpid.initial_config_virtualhost_config}", - "storeType": "DERBY" + "virtualHostInitialConfiguration": "{\"type\": \"Memory\"}" } ] } \ No newline at end of file
