diff --git a/settings.gradle b/settings.gradle index d0e8431f..fd330fe2 100644 --- a/settings.gradle +++ b/settings.gradle @@ -18,7 +18,7 @@ include 'spring-pulsar-cache-provider-caffeine' include 'spring-pulsar-reactive' include 'spring-pulsar-dependencies' include 'spring-pulsar-sample-apps:sample-imperative-produce-consume' -include 'spring-pulsar-sample-apps:sample-app2' +include 'spring-pulsar-sample-apps:sample-failover-custom-router' include 'spring-pulsar-sample-apps:sample-pulsar-functions:sample-signup-app' include 'spring-pulsar-sample-apps:sample-pulsar-functions:sample-signup-function' include 'spring-pulsar-sample-apps:sample-reactive' diff --git a/spring-pulsar-sample-apps/sample-app2/build.gradle b/spring-pulsar-sample-apps/sample-app2/build.gradle deleted file mode 100644 index 417305be..00000000 --- a/spring-pulsar-sample-apps/sample-app2/build.gradle +++ /dev/null @@ -1,35 +0,0 @@ -plugins { - id 'java' - id 'org.springframework.boot' version '3.2.0-SNAPSHOT' - id 'io.spring.dependency-management' version '1.1.0' -} - -description = 'Spring Pulsar Sample Applications (Custom Routing)' - -repositories { - mavenCentral() - maven { url 'https://repo.spring.io/milestone' } - maven { url 'https://repo.spring.io/snapshot' } -} - -dependencies { - implementation "org.springframework.boot:spring-boot-starter-pulsar:${springBootVersion}" - implementation 'org.springframework.boot:spring-boot-starter-actuator' - implementation 'io.micrometer:micrometer-tracing-bridge-brave' - implementation 'io.zipkin.reporter2:zipkin-reporter-brave' - implementation 'io.zipkin.reporter2:zipkin-sender-urlconnection' -} - -test { - useJUnitPlatform() - testLogging.showStandardStreams = true - outputs.upToDateWhen { false } -} - -bootRun { - jvmArgs = [ - "--add-opens", "java.base/java.lang=ALL-UNNAMED", - "--add-opens", "java.base/java.util=ALL-UNNAMED", - "--add-opens", "java.base/sun.net=ALL-UNNAMED" - ] -} diff --git a/spring-pulsar-sample-apps/sample-app2/src/main/java/app2/FailoverConsumerApp.java b/spring-pulsar-sample-apps/sample-app2/src/main/java/app2/FailoverConsumerApp.java deleted file mode 100644 index 15fe82d3..00000000 --- a/spring-pulsar-sample-apps/sample-app2/src/main/java/app2/FailoverConsumerApp.java +++ /dev/null @@ -1,114 +0,0 @@ -/* - * Copyright 2022-2023 the original author or authors. - * - * Licensed 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 - * - * https://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 app2; - -import java.io.Serial; - -import org.apache.pulsar.client.api.Message; -import org.apache.pulsar.client.api.MessageRouter; -import org.apache.pulsar.client.api.SubscriptionType; -import org.apache.pulsar.client.api.TopicMetadata; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.springframework.boot.ApplicationRunner; -import org.springframework.boot.SpringApplication; -import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.context.annotation.Bean; -import org.springframework.pulsar.annotation.PulsarListener; -import org.springframework.pulsar.core.PulsarTemplate; - -@SpringBootApplication -public class FailoverConsumerApp { - - private final Logger logger = LoggerFactory.getLogger(FailoverConsumerApp.class); - - public static void main(String[] args) { - SpringApplication.run(FailoverConsumerApp.class, "--spring.pulsar.producer.messageRoutingMode=CustomPartition"); - } - - @Bean - ApplicationRunner runner(PulsarTemplate pulsarTemplate) { - String topic = "failover-demo-topic"; - return args -> { - for (int i = 0; i < 10; i++) { - pulsarTemplate.newMessage("hello john doe 0 ").withTopic(topic) - .withProducerCustomizer(builder -> builder.messageRouter(new FooRouter())).sendAsync(); - pulsarTemplate.newMessage("hello alice doe 1").withTopic(topic) - .withProducerCustomizer(builder -> builder.messageRouter(new BarRouter())).sendAsync(); - pulsarTemplate.newMessage("hello buzz doe 2").withTopic(topic) - .withProducerCustomizer(builder -> builder.messageRouter(new BuzzRouter())).sendAsync(); - Thread.sleep(1_000); - } - }; - } - - @PulsarListener(subscriptionName = "failover-subscription-demo", topics = "failover-demo-topic", - subscriptionType = SubscriptionType.Failover) - void listen1(String foo) { - this.logger.info("failover-listen1 : " + foo); - } - - @PulsarListener(subscriptionName = "failover-subscription-demo", topics = "failover-demo-topic", - subscriptionType = SubscriptionType.Failover) - void listen2(String foo) { - this.logger.info("failover-listen2 : " + foo); - } - - @PulsarListener(subscriptionName = "failover-subscription-demo", topics = "failover-demo-topic", - subscriptionType = SubscriptionType.Failover) - void listen(String foo) { - this.logger.info("failover-listen3 : " + foo); - } - - static class FooRouter implements MessageRouter { - - @Serial - private static final long serialVersionUID = -1L; - - @Override - public int choosePartition(Message msg, TopicMetadata metadata) { - return 0; - } - - } - - static class BarRouter implements MessageRouter { - - @Serial - private static final long serialVersionUID = -1L; - - @Override - public int choosePartition(Message msg, TopicMetadata metadata) { - return 1; - } - - } - - static class BuzzRouter implements MessageRouter { - - @Serial - private static final long serialVersionUID = -1L; - - @Override - public int choosePartition(Message msg, TopicMetadata metadata) { - return 2; - } - - } - -} diff --git a/spring-pulsar-sample-apps/sample-app2/src/main/resources/application.yml b/spring-pulsar-sample-apps/sample-app2/src/main/resources/application.yml deleted file mode 100644 index 091b1570..00000000 --- a/spring-pulsar-sample-apps/sample-app2/src/main/resources/application.yml +++ /dev/null @@ -1,12 +0,0 @@ -logging: - level: - org.apache.pulsar: error - -management: - tracing: - enabled: false - sampling: - probability: 1.0 - zipkin: - tracing: - endpoint: "http://localhost:9411/api/v2/spans" diff --git a/spring-pulsar-sample-apps/sample-failover-custom-router/build.gradle b/spring-pulsar-sample-apps/sample-failover-custom-router/build.gradle new file mode 100644 index 00000000..bd7055c8 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-failover-custom-router/build.gradle @@ -0,0 +1,47 @@ +plugins { + id 'java' + id 'org.springframework.boot' version '3.2.1-SNAPSHOT' + id 'io.spring.dependency-management' version '1.1.4' +} + +description = 'Spring Pulsar Sample (Custom Routing w/ Failover)' + +repositories { + mavenCentral() + maven { url 'https://repo.spring.io/milestone' } + maven { url 'https://repo.spring.io/snapshot' } +} + +ext['spring-pulsar.version'] = '${project.version}' + +dependencies { + implementation 'org.springframework.boot:spring-boot-starter-pulsar' + implementation 'org.springframework.boot:spring-boot-starter-actuator' + implementation 'io.micrometer:micrometer-tracing-bridge-brave' + implementation 'io.zipkin.reporter2:zipkin-reporter-brave' + implementation 'io.zipkin.reporter2:zipkin-sender-urlconnection' + developmentOnly 'org.springframework.boot:spring-boot-docker-compose' + + testImplementation project(':spring-pulsar-test') + testRuntimeOnly 'ch.qos.logback:logback-classic' + testRuntimeOnly 'org.junit.platform:junit-platform-launcher' + testImplementation 'org.awaitility:awaitility' + testImplementation "org.springframework.boot:spring-boot-starter-test" + testImplementation "org.springframework.boot:spring-boot-testcontainers" + testImplementation 'org.testcontainers:junit-jupiter' + testImplementation 'org.testcontainers:pulsar' +} + +test { + useJUnitPlatform() + testLogging.showStandardStreams = true + outputs.upToDateWhen { false } +} + +bootRun { + jvmArgs = [ + "--add-opens", "java.base/java.lang=ALL-UNNAMED", + "--add-opens", "java.base/java.util=ALL-UNNAMED", + "--add-opens", "java.base/sun.net=ALL-UNNAMED" + ] +} diff --git a/spring-pulsar-sample-apps/sample-failover-custom-router/compose.yaml b/spring-pulsar-sample-apps/sample-failover-custom-router/compose.yaml new file mode 100644 index 00000000..013e7f4a --- /dev/null +++ b/spring-pulsar-sample-apps/sample-failover-custom-router/compose.yaml @@ -0,0 +1,7 @@ +services: + pulsar: + image: 'apachepulsar/pulsar:3.1.1' + ports: + - '6650' + - '8080' + command: 'bin/pulsar standalone' diff --git a/spring-pulsar-sample-apps/sample-failover-custom-router/src/main/java/com/example/FailoverConsumerApp.java b/spring-pulsar-sample-apps/sample-failover-custom-router/src/main/java/com/example/FailoverConsumerApp.java new file mode 100644 index 00000000..03d5551a --- /dev/null +++ b/spring-pulsar-sample-apps/sample-failover-custom-router/src/main/java/com/example/FailoverConsumerApp.java @@ -0,0 +1,105 @@ +/* + * Copyright 2022-2023 the original author or authors. + * + * Licensed 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 + * + * https://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 com.example; + +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.MessageRouter; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.client.api.TopicMetadata; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.springframework.boot.ApplicationRunner; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.Bean; +import org.springframework.pulsar.annotation.PulsarListener; +import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.core.PulsarTopic; + +@SpringBootApplication +public class FailoverConsumerApp { + + private static final Logger LOG = LoggerFactory.getLogger(FailoverConsumerApp.class); + + private static final String TOPIC = "failover-demo-topic"; + + public static void main(String[] args) { + SpringApplication.run(FailoverConsumerApp.class, args); + } + + @Bean + PulsarTopic failoverDemoTopic() { + return PulsarTopic.builder(TOPIC).numberOfPartitions(3).build(); + } + + @Bean + ApplicationRunner runner(PulsarTemplate template) { + return (args) -> { + for (int i = 0; i < 10; i++) { + sendMessage(0, template, new PartitionZeroRouter()); + sendMessage(1, template, new PartitionOneRouter()); + sendMessage(2, template, new PartitionTwoRouter()); + } + }; + } + + private void sendMessage(int partition, PulsarTemplate template, MessageRouter router) throws PulsarClientException { + var msg = "hello_" + partition; + template.newMessage(msg).withTopic(TOPIC) + .withProducerCustomizer(builder -> builder.messageRouter(router)).sendAsync(); + LOG.info("++++++PRODUCE_{} {}------", partition, msg); + } + + @PulsarListener(topics = TOPIC, subscriptionName = TOPIC+"-sub", subscriptionType = SubscriptionType.Failover) + void listen0(String msg) { + LOG.info("++++++CONSUME_0 {}------", msg); + } + + @PulsarListener(topics = TOPIC, subscriptionName = TOPIC+"-sub", subscriptionType = SubscriptionType.Failover) + void listen1(String msg) { + LOG.info("++++++CONSUME_1 {}------", msg); + } + + @PulsarListener(topics = TOPIC, subscriptionName = TOPIC+"-sub", subscriptionType = SubscriptionType.Failover) + void listen2(String msg) { + LOG.info("++++++CONSUME_2 {}------", msg); + } + + static class PartitionZeroRouter implements MessageRouter { + @Override + public int choosePartition(Message msg, TopicMetadata metadata) { + return 0; + } + } + + static class PartitionOneRouter implements MessageRouter { + @Override + public int choosePartition(Message msg, TopicMetadata metadata) { + return 1; + } + } + + static class PartitionTwoRouter implements MessageRouter { + @Override + public int choosePartition(Message msg, TopicMetadata metadata) { + return 2; + } + } + +} diff --git a/spring-pulsar-sample-apps/sample-app2/src/main/java/app2/package-info.java b/spring-pulsar-sample-apps/sample-failover-custom-router/src/main/java/com/example/package-info.java similarity index 89% rename from spring-pulsar-sample-apps/sample-app2/src/main/java/app2/package-info.java rename to spring-pulsar-sample-apps/sample-failover-custom-router/src/main/java/com/example/package-info.java index 552340e1..5f2a90d1 100644 --- a/spring-pulsar-sample-apps/sample-app2/src/main/java/app2/package-info.java +++ b/spring-pulsar-sample-apps/sample-failover-custom-router/src/main/java/com/example/package-info.java @@ -3,7 +3,7 @@ */ @NonNullApi @NonNullFields -package app2; +package com.example; import org.springframework.lang.NonNullApi; import org.springframework.lang.NonNullFields; diff --git a/spring-pulsar-sample-apps/sample-failover-custom-router/src/main/resources/application.yml b/spring-pulsar-sample-apps/sample-failover-custom-router/src/main/resources/application.yml new file mode 100644 index 00000000..7267f11a --- /dev/null +++ b/spring-pulsar-sample-apps/sample-failover-custom-router/src/main/resources/application.yml @@ -0,0 +1,7 @@ +spring: + pulsar: + producer: + message-routing-mode: custompartition + docker: + compose: + file: spring-pulsar-sample-apps/sample-failover-custom-router/compose.yaml diff --git a/spring-pulsar-sample-apps/sample-failover-custom-router/src/test/java/com/example/FailoverConsumerAppTests.java b/spring-pulsar-sample-apps/sample-failover-custom-router/src/test/java/com/example/FailoverConsumerAppTests.java new file mode 100644 index 00000000..cfbef2d7 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-failover-custom-router/src/test/java/com/example/FailoverConsumerAppTests.java @@ -0,0 +1,79 @@ +/* + * Copyright 2012-2023 the original author or authors. + * + * Licensed 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 + * + * https://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 com.example; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.regex.Pattern; +import java.util.stream.IntStream; + +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.system.CapturedOutput; +import org.springframework.boot.test.system.OutputCaptureExtension; +import org.springframework.pulsar.test.support.PulsarTestContainerSupport; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; + +@SpringBootTest +@ExtendWith(OutputCaptureExtension.class) +class FailoverConsumerAppTests implements PulsarTestContainerSupport { + + @DynamicPropertySource + static void pulsarProperties(DynamicPropertyRegistry registry) { + registry.add("spring.pulsar.client.service-url", PULSAR_CONTAINER::getPulsarBrokerUrl); + registry.add("spring.pulsar.admin.service-url", PULSAR_CONTAINER::getHttpServiceUrl); + } + + @Test + void produceConsumeWithPrimitiveMessageType(CapturedOutput output) { + List expectedOutput = new ArrayList<>(); + IntStream.range(0, 10).forEachOrdered((i) -> { + expectedOutput.add("++++++PRODUCE_0 hello_0------"); + expectedOutput.add("++++++PRODUCE_1 hello_1------"); + expectedOutput.add("++++++PRODUCE_2 hello_2------"); + }); + Awaitility.waitAtMost(Duration.ofSeconds(15)) + .untilAsserted(() -> assertThat(output).satisfies((out) -> { + assertThat(output).contains(expectedOutput); + assertListenerConsumedNumMessagesFromSinglePartitionOnly(0, out.toString()); + assertListenerConsumedNumMessagesFromSinglePartitionOnly(1, out.toString()); + assertListenerConsumedNumMessagesFromSinglePartitionOnly(2, out.toString()); + })); + } + + private void assertListenerConsumedNumMessagesFromSinglePartitionOnly(int listenerIndex, String output) { + var msgsConsumedByPartition = new HashMap<>(); + msgsConsumedByPartition.put(0, numMessagesConsumed(listenerIndex, 0, output)); + msgsConsumedByPartition.put(1, numMessagesConsumed(listenerIndex, 1, output)); + msgsConsumedByPartition.put(2, numMessagesConsumed(listenerIndex, 2, output)); + var numMatched = msgsConsumedByPartition.values().stream().filter(Long.valueOf(10)::equals).count(); + assertThat(numMatched).isEqualTo(1); + } + + private long numMessagesConsumed(int consumerPartition, int producerPartition, String output) { + var regex = "(\\+\\+\\+\\+\\+\\+CONSUME_%d hello_%d------)".formatted(consumerPartition, producerPartition); + return Pattern.compile(regex).matcher(output).results().count(); + } +} diff --git a/spring-pulsar-sample-apps/sample-failover-custom-router/src/test/resources/logback-test.xml b/spring-pulsar-sample-apps/sample-failover-custom-router/src/test/resources/logback-test.xml new file mode 100644 index 00000000..97f7e370 --- /dev/null +++ b/spring-pulsar-sample-apps/sample-failover-custom-router/src/test/resources/logback-test.xml @@ -0,0 +1,14 @@ + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n + + + + + + + + + +