Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Convert samples to tests (#504)

Browse Source 
* Add tests for FailoverConsumerApp sample app

* Renamed package for FailoverConsumerApp to `com.example` from `app2`
This commit is contained in:
Chris Bono
2023-11-26 23:05:05 -06:00
committed by GitHub
parent 6155fd593b
commit 3aa41e2107
11 changed files with 261 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
settings.gradle
View File

@@ -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'

35
spring-pulsar-sample-apps/sample-app2/build.gradle
View File

@@ -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"
]
}

114
spring-pulsar-sample-apps/sample-app2/src/main/java/app2/FailoverConsumerApp.java
View File

@@ -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<String> 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;
}
}
}

12
spring-pulsar-sample-apps/sample-app2/src/main/resources/application.yml
View File

@@ -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"

47
spring-pulsar-sample-apps/sample-failover-custom-router/build.gradle Normal file
View File

@@ -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"
]
}

7
spring-pulsar-sample-apps/sample-failover-custom-router/compose.yaml Normal file
View File

@@ -0,0 +1,7 @@
services:
pulsar:
image: 'apachepulsar/pulsar:3.1.1'
ports:
- '6650'
- '8080'
command: 'bin/pulsar standalone'

105
spring-pulsar-sample-apps/sample-failover-custom-router/src/main/java/com/example/FailoverConsumerApp.java Normal file
View File

@@ -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<String> 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<String> 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;
}
}
}

2
spring-pulsar-sample-apps/sample-app2/src/main/java/app2/package-info.java → spring-pulsar-sample-apps/sample-failover-custom-router/src/main/java/com/example/package-info.java
View File

@@ -3,7 +3,7 @@
*/
@NonNullApi
@NonNullFields
package app2;
package com.example;
import org.springframework.lang.NonNullApi;
import org.springframework.lang.NonNullFields;

7
spring-pulsar-sample-apps/sample-failover-custom-router/src/main/resources/application.yml Normal file
View File

@@ -0,0 +1,7 @@
spring:
pulsar:
producer:
message-routing-mode: custompartition
docker:
compose:
file: spring-pulsar-sample-apps/sample-failover-custom-router/compose.yaml

79
spring-pulsar-sample-apps/sample-failover-custom-router/src/test/java/com/example/FailoverConsumerAppTests.java Normal file
View File

@@ -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 <String> 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();
}
}

14
spring-pulsar-sample-apps/sample-failover-custom-router/src/test/resources/logback-test.xml Normal file
View File

@@ -0,0 +1,14 @@
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n</pattern>
</encoder>
</appender>
<root level="WARN">
<appender-ref ref="STDOUT"/>
</root>
<logger name="com.example" level="INFO"/>
<logger name="com.github.dockerjava" level="ERROR"/>
<logger name="org.apache.pulsar.common.util.netty" level="ERROR" />
<logger name="org.testcontainers" level="ERROR"/>
</configuration>