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

Add test for imperative consumer properties (#220)

Browse Source
This commit is contained in:
Christophe Bornet
2022-11-18 23:03:15 +01:00
committed by GitHub
parent 0e4a4fad38
commit 6f2995e1b3
1 changed files with 70 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java
View File

@@ -19,15 +19,20 @@ package org.springframework.pulsar.autoconfigure;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
import org.apache.pulsar.client.api.HashingScheme;
import org.apache.pulsar.client.api.MessageRoutingMode;
import org.apache.pulsar.client.api.ProducerAccessMode;
import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
import org.apache.pulsar.client.api.RegexSubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionType;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
@@ -58,11 +63,11 @@ public class PulsarPropertiesTests {
@Nested
class AdminPropertiesTests {
private String authPluginClassName = "org.apache.pulsar.client.impl.auth.AuthenticationToken";
private final String authPluginClassName = "org.apache.pulsar.client.impl.auth.AuthenticationToken";
private String authParamsStr = "{\"token\":\"1234\"}";
private final String authParamsStr = "{\"token\":\"1234\"}";
private String authToken = "1234";
private final String authToken = "1234";
@Test
void authenticationUsingAuthParamsString() {
@@ -98,9 +103,8 @@ public class PulsarPropertiesTests {
props.put("spring.pulsar.administration.auth-params", authParamsStr);
props.put("spring.pulsar.administration.authentication.token", authToken);
bind(props);
assertThatIllegalArgumentException().isThrownBy(() -> properties.buildAdminProperties())
.withMessageContaining(
"Cannot set both spring.pulsar.administration.authParams and spring.pulsar.administration.authentication.*");
assertThatIllegalArgumentException().isThrownBy(properties::buildAdminProperties).withMessageContaining(
"Cannot set both spring.pulsar.administration.authParams and spring.pulsar.administration.authentication.*");
}
}
@@ -145,4 +149,64 @@ public class PulsarPropertiesTests {
}
@Nested
class ConsumerPropertiesTests {
@Test
@SuppressWarnings("unchecked")
void consumerProperties() {
Map<String, String> props = new HashMap<>();
props.put("spring.pulsar.consumer.topics[0]", "my-topic");
props.put("spring.pulsar.consumer.topics-pattern", "my-pattern");
props.put("spring.pulsar.consumer.subscription-name", "my-subscription");
props.put("spring.pulsar.consumer.subscription-type", "Shared");
props.put("spring.pulsar.consumer.receiver-queue-size", "1");
props.put("spring.pulsar.consumer.acknowledgements-group-time", "2s");
props.put("spring.pulsar.consumer.negative-ack-redelivery-delay", "3s");
props.put("spring.pulsar.consumer.max-total-receiver-queue-size-across-partitions", "5");
props.put("spring.pulsar.consumer.consumer-name", "my-consumer");
props.put("spring.pulsar.consumer.ack-timeout", "6s");
props.put("spring.pulsar.consumer.tick-duration", "7s");
props.put("spring.pulsar.consumer.priority-level", "8");
props.put("spring.pulsar.consumer.crypto-failure-action", "DISCARD");
props.put("spring.pulsar.consumer.properties[my-prop]", "my-prop-value");
props.put("spring.pulsar.consumer.read-compacted", "true");
props.put("spring.pulsar.consumer.subscription-initial-position", "Earliest");
props.put("spring.pulsar.consumer.pattern-auto-discovery-period", "9");
props.put("spring.pulsar.consumer.regex-subscription-mode", "AllTopics");
props.put("spring.pulsar.consumer.auto-update-partitions", "false");
props.put("spring.pulsar.consumer.replicate-subscription-state", "true");
props.put("spring.pulsar.consumer.auto-ack-oldest-chunked-message-on-queue-full", "false");
props.put("spring.pulsar.consumer.max-pending-chunked-message", "11");
props.put("spring.pulsar.consumer.expire-time-of-incomplete-chunked-message", "12s");
bind(props);
Map<String, Object> consumerProps = properties.buildConsumerProperties();
assertThat(consumerProps)
.hasEntrySatisfying("topicNames",
n -> assertThat((Collection<String>) n).containsExactly("my-topic"))
.hasEntrySatisfying("topicsPattern", p -> assertThat(p.toString()).isEqualTo("my-pattern"))
.containsEntry("subscriptionName", "my-subscription")
.containsEntry("subscriptionType", SubscriptionType.Shared).containsEntry("receiverQueueSize", 1)
.containsEntry("acknowledgementsGroupTimeMicros", 2_000_000L)
.containsEntry("negativeAckRedeliveryDelayMicros", 3_000_000L)
.containsEntry("maxTotalReceiverQueueSizeAcrossPartitions", 5)
.containsEntry("consumerName", "my-consumer").containsEntry("ackTimeoutMillis", 6_000L)
.containsEntry("tickDurationMillis", 7_000L).containsEntry("priorityLevel", 8)
.containsEntry("cryptoFailureAction", ConsumerCryptoFailureAction.DISCARD)
.hasEntrySatisfying("properties",
p -> assertThat((Map<String, String>) p).containsEntry("my-prop", "my-prop-value"))
.containsEntry("readCompacted", true)
.containsEntry("subscriptionInitialPosition", SubscriptionInitialPosition.Earliest)
.containsEntry("patternAutoDiscoveryPeriod", 9)
.containsEntry("regexSubscriptionMode", RegexSubscriptionMode.AllTopics)
.containsEntry("autoUpdatePartitions", false).containsEntry("replicateSubscriptionState", true)
.containsEntry("autoAckOldestChunkedMessageOnQueueFull", false)
.containsEntry("maxPendingChunkedMessage", 11)
.containsEntry("expireTimeOfIncompleteChunkedMessageMillis", 12_000L);
}
}
}