From 67cb57c800a8cca0f6bc460ddd66af34c02ebcfa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Preu=C3=9F?= Date: Mon, 29 Aug 2022 20:30:16 +0200 Subject: [PATCH] Add auto-configuration for PulsarAdministration (#72) * Auto-configures a `PulsarAdministration` * Reconciles some properties for admin and regular clients * Adds docs for config props (see #50) Resolves #41 --- .../DocumentConfigurationProperties.java | 2 + .../main/asciidoc/application-properties.adoc | 2 + .../src/main/asciidoc/pulsar.adoc | 36 + .../main/java/app1/SpringPulsarBootApp.java | 32 +- .../PulsarAutoConfiguration.java | 7 + .../autoconfigure/PulsarProperties.java | 800 +++++++++++++++++- .../PulsarAutoConfigurationTests.java | 12 +- .../pulsar/core/PulsarAdministration.java | 26 +- .../pulsar/core/PulsarTopic.java | 3 +- .../listener/PulsarContainerProperties.java | 2 +- .../core/PulsarAdministrationTests.java | 4 +- 11 files changed, 861 insertions(+), 65 deletions(-) diff --git a/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java b/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java index 5bcb2dbd..86a9a838 100644 --- a/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java +++ b/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java @@ -34,6 +34,7 @@ import org.gradle.api.tasks.TaskAction; * @author Andy Wilkinson * @author Phillip Webb * @author Chris Bono + * @author Alexander Preuß */ public class DocumentConfigurationProperties extends DefaultTask { @@ -69,6 +70,7 @@ public class DocumentConfigurationProperties extends DefaultTask { c.accept("spring.pulsar.consumer"); c.accept("spring.pulsar.listener"); }); + snippets.add("application-properties.pulsar-administration", "Pulsar Administration Properties", (c) -> c.accept("spring.pulsar.administration")); snippets.writeTo(this.outputDir.toPath()); } } diff --git a/spring-pulsar-docs/src/main/asciidoc/application-properties.adoc b/spring-pulsar-docs/src/main/asciidoc/application-properties.adoc index 0f94d9d9..25dea9f5 100644 --- a/spring-pulsar-docs/src/main/asciidoc/application-properties.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/application-properties.adoc @@ -15,3 +15,5 @@ include::application-properties/pulsar-client.adoc[] include::application-properties/pulsar-producer.adoc[] include::application-properties/pulsar-consumer.adoc[] + +include::application-properties/pulsar-administration.adoc[] diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc index 3284dd3f..f8797d0e 100644 --- a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc @@ -571,6 +571,42 @@ ProducerInterceptor secondInterceptor() { ---- ==== +[[pulsar-admin]] +==== Pulsar Admin +On the Pulsar administration side, Spring Boot auto-configuration provides a `PulsarAdministration` to manage Pulsar clusters. +The administration implements an interface called `PulsarAdminOperations` and provides {javadocs}/org/springframework/pulsar/core/PulsarAdminOperations.html[a 'createOrModify' method] to handle topic administration through its contract. + +When using the Pulsar Spring Boot Starter, you get the `PulsarAdministration` auto-configured. +By default, the application tries to connect to a local Pulsar instance at `http://localhost:8080`. However, there are many application properties available to configure the client. + +.[.underline]#Click ##here## to view the available **Pulsar Administration Properties**#. +[%collapsible] +==== +include::application-properties/pulsar-administration.adoc[lines=3..-1] +==== + +On initialization, the `PulsarAdministration` checks if there are any `PulsarTopic` beans in the application context. +For all such beans, the `PulsarAdministration` will either create the corresponding topic, or if necessary modify the number of partitions. + +Below is an example how to add `PulsarTopic` beans to let the `PulsarAdministration` auto-create topics for you. + +==== +[source,java] +---- +@Bean +PulsarTopic simpleTopic { + // This will create a non-partitioned topic in the public/default namespace + return PulsarTopic.builder("simple-topic").build(); +} + +@Bean +PulsarTopic partitionedTopic { + // This will create a partitioned topic with 3 partitions in the provided tenant and namespace + return PulsarTopic.builder("persistent://my-tenant/my-namespace/partitioned-topic", 3).build(); +} +---- +==== + ==== Appendix The reference documentation has the following appendices: diff --git a/spring-pulsar-sample-apps/src/main/java/app1/SpringPulsarBootApp.java b/spring-pulsar-sample-apps/src/main/java/app1/SpringPulsarBootApp.java index e45f380c..25432b51 100644 --- a/spring-pulsar-sample-apps/src/main/java/app1/SpringPulsarBootApp.java +++ b/spring-pulsar-sample-apps/src/main/java/app1/SpringPulsarBootApp.java @@ -31,6 +31,7 @@ import org.springframework.context.annotation.Bean; import org.springframework.pulsar.annotation.PulsarListener; import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.core.PulsarTopic; @SpringBootApplication public class SpringPulsarBootApp { @@ -107,13 +108,36 @@ public class SpringPulsarBootApp { this.logger.info("Message received :" + message); } + /* + * Create a partitioned topic using PulsarAdministration and then publish to the topic + * and consume from it. + */ + @Bean + PulsarTopic partitionedTopic4() { + return PulsarTopic.builder("hello-pulsar-partitioned-4").numberOfPartitions(3).build(); + } + + @Bean + ApplicationRunner runner4(PulsarTemplate pulsarTemplate) { + return args -> { + for (int i = 0; i < 10; i++) { + pulsarTemplate.send("hello-pulsar-partitioned-4", "This is message " + (i + 1)); + } + }; + } + + @PulsarListener(subscriptionName = "subscription-4", topics = "hello-pulsar-partitioned-4") + void listen4(String message) { + this.logger.info("Message received from partitioned-topic : " + message); + } + /* * Publish and then use PulsarListener in batch listening mode. */ @Bean - ApplicationRunner runner4(PulsarProducerFactory producerFactory) { + ApplicationRunner runner5(PulsarProducerFactory producerFactory) { - String topic = "hello-pulsar-exclusive-4"; + String topic = "hello-pulsar-exclusive-5"; PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory); pulsarTemplate.setSchema(Schema.JSON(Foo.class)); return args -> { @@ -124,9 +148,9 @@ public class SpringPulsarBootApp { }; } - @PulsarListener(subscriptionName = "subscription-4", topics = "hello-pulsar-exclusive-4", + @PulsarListener(subscriptionName = "subscription-5", topics = "hello-pulsar-exclusive-5", schemaType = SchemaType.JSON, batch = true) - void listen4(List messages) { + void listen5(List messages) { this.logger.info("records received :" + messages.size()); for (Foo message : messages) { this.logger.info("record : " + message); diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java index 39fdd85b..543c037d 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java @@ -33,6 +33,7 @@ import org.springframework.pulsar.config.PulsarClientFactoryBean; import org.springframework.pulsar.core.CachingPulsarProducerFactory; import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; @@ -98,4 +99,10 @@ public class PulsarAutoConfiguration { return new DefaultPulsarConsumerFactory<>(pulsarClient, this.properties.buildConsumerProperties()); } + @Bean + @ConditionalOnMissingBean(PulsarAdministration.class) + public PulsarAdministration pulsarAdministration() { + return new PulsarAdministration(this.properties.buildAdminProperties()); + } + } diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java index 03e71e8e..774c79a3 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java @@ -47,6 +47,7 @@ import org.springframework.pulsar.listener.PulsarContainerProperties; * properties. * * @author Soby Chacko + * @author Alexander Preuß */ @ConfigurationProperties(prefix = "spring.pulsar") public class PulsarProperties { @@ -59,80 +60,167 @@ public class PulsarProperties { private final Producer producer = new Producer(); - public Map buildConsumerProperties() { - return new HashMap<>(this.consumer.buildProperties()); - } - - public Map buildProducerProperties() { - return new HashMap<>(this.producer.buildProperties()); - } + private final Admin admin = new Admin(); public Consumer getConsumer() { return this.consumer; } - public Listener getListener() { - return this.listener; - } - public Client getClient() { return this.client; } + public Listener getListener() { + return this.listener; + } + public Producer getProducer() { return this.producer; } + public Admin getAdministration() { + return this.admin; + } + + public Map buildConsumerProperties() { + return new HashMap<>(this.consumer.buildProperties()); + } + public Map buildClientProperties() { return new HashMap<>(this.client.buildProperties()); } + public Map buildProducerProperties() { + return new HashMap<>(this.producer.buildProperties()); + } + + public Map buildAdminProperties() { + return new HashMap<>(this.admin.buildProperties()); + } + public static class Consumer { + /** + * Comma-separated list of topics the consumer subscribes to. + */ private String[] topics; + /** + * Pattern for topics the consumer subscribes to. + */ private String topicsPattern; + /** + * Subscription name for the consumer. + */ private String subscriptionName; + /** + * Subscription type to be used when subscribing to a topic. + */ private SubscriptionType subscriptionType = SubscriptionType.Exclusive; + /** + * Number of messages that can be accumulated before the consumer calls "receive". + */ private int receiverQueueSize = 1000; + /** + * Time to group acknowledgements before sending them to the broker in + * microseconds. + */ private long acknowledgementsGroupTimeMicros = TimeUnit.MILLISECONDS.toMicros(100); + /** + * Delay before re-delivering messages that have failed to be processed in + * microseconds. + */ private long negativeAckRedeliveryDelayMicros = TimeUnit.MINUTES.toMicros(1); + /** + * Maximum number of messages that a consumer can be pushed at once from a broker + * across all partitions. + */ private int maxTotalReceiverQueueSizeAcrossPartitions = 50000; + /** + * Consumer name to identify a particular consumer from the topic stats. + */ private String consumerName; + /** + * Timeout for unacked messages to be redelivered. + */ private long ackTimeoutMillis = 0; + /** + * Precision for the ack timeout messages tracker in milliseconds. + */ private long tickDurationMillis = 1000; + /** + * Priority level for shared subscription consumers. + */ private int priorityLevel = 0; + /** + * Action the consumer will take in case of decryption failure. + */ private ConsumerCryptoFailureAction cryptoFailureAction = ConsumerCryptoFailureAction.FAIL; + /** + * Map of properties to add to the consumer. + */ private SortedMap properties = new TreeMap<>(); + /** + * Whether to read messages from the compacted topic rather than the full message + * backlog. + */ private boolean readCompacted = false; + /** + * Position where to initialize a newly created subscription. + */ private SubscriptionInitialPosition subscriptionInitialPosition = SubscriptionInitialPosition.Latest; + /** + * Auto-discovery period for topics when topic pattern is used in minutes. + */ private int patternAutoDiscoveryPeriod = 1; + /** + * Determines which topics the consumer should be subscribed to when using pattern + * subscriptions. + */ private RegexSubscriptionMode regexSubscriptionMode = RegexSubscriptionMode.PersistentOnly; + /** + * Whether the consumer auto-subscribes for partition increase. This is only for + * partitioned consumers. + */ private boolean autoUpdatePartitions = true; + /** + * Whether to replicate subscription state. + */ private boolean replicateSubscriptionState = false; + /** + * Whether to automatically drop outstanding un-acked messages if the queue is + * full. + */ private boolean autoAckOldestChunkedMessageOnQueueFull = true; + /** + * Maximum number of chunked messages to be kept in memory. + */ private int maxPendingChunkedMessage = 10; + /** + * Time to expire incomplete chunks if the consumer won't be able to receive all + * chunks before in milliseconds. + */ private long expireTimeOfIncompleteChunkedMessageMillis = 60000; public String[] getTopics() { @@ -357,36 +445,86 @@ public class PulsarProperties { public static class Producer { + /** + * Topic the producer will publish to. + */ private String topicName; + /** + * Name for the producer. If not assigned, a unique name is generated. + */ private String producerName; + /** + * Time before a message has to be acknowledged by the broker in milliseconds. + */ private long sendTimeoutMs = 30000; + /** + * Whether the "send" and "sendAsync" methods should block if the outgoing message + * queue is full. + */ private boolean blockIfQueueFull = false; + /** + * Maximum number of pending messages for the producer. + */ private int maxPendingMessages = 1000; + /** + * Maximum number of pending messages across all the partitions. + */ private int maxPendingMessagesAcrossPartitions = 50000; + /** + * Message routing mode for a partitioned producer. + */ private MessageRoutingMode messageRoutingMode = MessageRoutingMode.RoundRobinPartition; + /** + * Message hashing scheme to choose the partition to which the message is + * published. + */ private HashingScheme hashingScheme = HashingScheme.JavaStringHash; + /** + * Action the producer will take in case of encryption failure. + */ private ProducerCryptoFailureAction cryptoFailureAction = ProducerCryptoFailureAction.FAIL; + /** + * Time period within which the messages sent will be batched in milliseconds. + */ private long batchingMaxPublishDelayMicros = TimeUnit.MILLISECONDS.toMicros(1); + /** + * Maximum number of messages to be batched. + */ private int batchingMaxMessages = 1000; + /** + * Whether to automatically batch messages. + */ private boolean batchingEnabled = true; + /** + * Whether to split large-size messages into multiple chunks. + */ private boolean chunkingEnabled = false; + /** + * Message compression type. + */ private CompressionType compressionType; + /** + * Name of the initial subscription of the topic. + */ private String initialSubscriptionName; + /** + * Type of access to the topic the producer requires. + */ private ProducerAccessMode producerAccessMode = ProducerAccessMode.Shared; private Cache cache = new Cache(); @@ -590,44 +728,218 @@ public class PulsarProperties { public static class Client { + /** + * Pulsar cluster URL to connect to a broker. + */ private String serviceUrl; + /** + * Listener name for lookup. Clients can use listenerName to choose one of the + * listeners as the service URL to create a connection to the broker. To use this, + * "advertisedListeners" must be enabled on the broker. + */ + private String listenerName; + + /** + * Class name of the clients' authentication plugin. + */ private String authPluginClassName; + /** + * Authentication parameter(s) of the client. + */ private String authParams; + /** + * Authentication parameter map of the client. + */ + private Map authParamsMap; + + /** + * Client operation timeout in milliseconds. + */ private long operationTimeoutMs = 30000L; - private long statsIntervalSeconds = 60; + /** + * Client lookup timeout in milliseconds. + */ + private long lookupTimeoutMs = -1; + /** + * Number of threads to be used for handling connections to brokers. + */ private int numIoThreads = 1; + /** + * Number of threads to be used for message listeners. + * + * The listener thread pool is shared across all the consumers and readers that + * are using a "listener" model to get messages. For a given consumer, the + * listener will always be invoked from the same thread, to ensure ordering. + */ + private int numListenerThreads = 1; + + /** + * Maximum number of connections that the client will open to a single broker. + */ + private int numConnectionsPerBroker = 1; + + /** + * Whether to use TCP no-delay flag on the connection, to disable Nagle algorithm. + */ private boolean useTcpNoDelay = true; + /** + * Whether to use TLS encryption on the connection. + */ private boolean useTls = false; - private String tlsTrustCertsFilePath; - - private boolean tlsAllowInsecureConnection = false; - + /** + * Whether the hostname is validated when the proxy creates a TLS connection with + * brokers. + */ private boolean tlsHostnameVerificationEnable = false; - private int concurrentLookupRequest = 5000; + /** + * Path to the trusted TLS certificate file. + */ + private String tlsTrustCertsFilePath; + /** + * Whether the client accepts untrusted TLS certificates from the broker. + */ + private boolean tlsAllowInsecureConnection = false; + + /** + * Enable KeyStore instead of PEM type configuration if TLS is enabled. + */ + private boolean useKeyStoreTls = false; + + /** + * Name of the security provider used for SSL connections. + */ + private String sslProvider; + + /** + * File format of the trust store file. + */ + private String tlsTrustStoreType; + + /** + * Location of the trust store file. + */ + private String tlsTrustStorePath; + + /** + * Store password for the key store file. + */ + private String tlsTrustStorePassword; + + /** + * Comma-separated list of cipher suites. This is a named combination of + * authentication, encryption, MAC and key exchange algorithm used to negotiate + * the security settings for a network connection using TLS or SSL network + * protocol. By default, all the available cipher suites are supported. + */ + private String[] tlsCiphers; + + /** + * Comma-separated list of SSL protocols used to generate the SSLContext. Allowed + * values in recent JVMs are TLS, TLSv1.3, TLSv1.2 and TLSv1.1. + */ + private String[] tlsProtocols; + + /** + * Interval between each stat info in seconds. + */ + private long statsIntervalSeconds = 60; + + /** + * Number of concurrent lookup-requests allowed to send on each broker-connection + * to prevent overload on broker. + */ + private int maxConcurrentLookupRequest = 5000; + + /** + * Number of max lookup-requests allowed on each broker-connection to prevent + * overload on broker. + */ private int maxLookupRequest = 50000; + /** + * Maximum number of times a lookup-request to a broker will be redirected. + */ + private int maxLookupRedirects = 20; + + /** + * Maximum number of broker-rejected requests in a certain timeframe, after which + * the current connection is closed and a new connection is created by the client. + */ private int maxNumberOfRejectedRequestPerConnection = 50; + /** + * Keep alive interval for broker-client connection in seconds. + */ private int keepAliveIntervalSeconds = 30; + /** + * Duration to wait for a connection to a broker to be established in + * milliseconds. + */ private int connectionTimeoutMs = 10000; - private int requestTimeoutMs = 60000; - + /** + * Initial backoff interval in nanoseconds. + */ private long initialBackoffIntervalNanos = TimeUnit.MILLISECONDS.toNanos(100); + /** + * Maximum backoff interval in nanoseconds. + */ private long maxBackoffIntervalNanos = TimeUnit.SECONDS.toNanos(30); + /** + * Enables spin-waiting on executors and IO threads in order to reduce latency + * during context switches. + */ + private boolean enableBusyWait = false; + + /** + * Limit of direct memory that will be allocated by the client. + */ + private long memoryLimitBytes = 64 * 1024 * 1024; + + /** + * Enables transactions. To use this, start the transactionCoordinatorClient with + * the pulsar client. + */ + private boolean enableTransaction = false; + + /** + * DNS lookup bind address. + */ + private String dnsLookupBindAddress; + + /** + * DNS lookup bind port. + */ + private int dnsLookupBindPort = 0; + + /** + * SOCKS5 proxy address. + */ + private String socks5ProxyAddress; + + /** + * SOCKS5 proxy username. + */ + private String socks5ProxyUsername; + + /** + * SOCKS5 proxy password. + */ + private String socks5ProxyPassword; + public String getServiceUrl() { return this.serviceUrl; } @@ -636,6 +948,14 @@ public class PulsarProperties { this.serviceUrl = serviceUrl; } + public String getListenerName() { + return this.listenerName; + } + + public void setListenerName(String listenerName) { + this.listenerName = listenerName; + } + public String getAuthPluginClassName() { return this.authPluginClassName; } @@ -652,6 +972,14 @@ public class PulsarProperties { this.authParams = authParams; } + public Map getAuthParamsMap() { + return this.authParamsMap; + } + + public void setAuthParamsMap(Map authParamsMap) { + this.authParamsMap = authParamsMap; + } + public long getOperationTimeoutMs() { return this.operationTimeoutMs; } @@ -660,12 +988,12 @@ public class PulsarProperties { this.operationTimeoutMs = operationTimeoutMs; } - public long getStatsIntervalSeconds() { - return this.statsIntervalSeconds; + public long getLookupTimeoutMs() { + return this.lookupTimeoutMs; } - public void setStatsIntervalSeconds(long statsIntervalSeconds) { - this.statsIntervalSeconds = statsIntervalSeconds; + public void setLookupTimeoutMs(long lookupTimeoutMs) { + this.lookupTimeoutMs = lookupTimeoutMs; } public int getNumIoThreads() { @@ -676,6 +1004,22 @@ public class PulsarProperties { this.numIoThreads = numIoThreads; } + public int getNumListenerThreads() { + return this.numListenerThreads; + } + + public void setNumListenerThreads(int numListenerThreads) { + this.numListenerThreads = numListenerThreads; + } + + public int getNumConnectionsPerBroker() { + return this.numConnectionsPerBroker; + } + + public void setNumConnectionsPerBroker(int numConnectionsPerBroker) { + this.numConnectionsPerBroker = numConnectionsPerBroker; + } + public boolean isUseTcpNoDelay() { return this.useTcpNoDelay; } @@ -692,6 +1036,14 @@ public class PulsarProperties { this.useTls = useTls; } + public boolean isTlsHostnameVerificationEnable() { + return this.tlsHostnameVerificationEnable; + } + + public void setTlsHostnameVerificationEnable(boolean tlsHostnameVerificationEnable) { + this.tlsHostnameVerificationEnable = tlsHostnameVerificationEnable; + } + public String getTlsTrustCertsFilePath() { return this.tlsTrustCertsFilePath; } @@ -708,20 +1060,76 @@ public class PulsarProperties { this.tlsAllowInsecureConnection = tlsAllowInsecureConnection; } - public boolean isTlsHostnameVerificationEnable() { - return this.tlsHostnameVerificationEnable; + public boolean isUseKeyStoreTls() { + return this.useKeyStoreTls; } - public void setTlsHostnameVerificationEnable(boolean tlsHostnameVerificationEnable) { - this.tlsHostnameVerificationEnable = tlsHostnameVerificationEnable; + public void setUseKeyStoreTls(boolean useKeyStoreTls) { + this.useKeyStoreTls = useKeyStoreTls; } - public int getConcurrentLookupRequest() { - return this.concurrentLookupRequest; + public String getSslProvider() { + return this.sslProvider; } - public void setConcurrentLookupRequest(int concurrentLookupRequest) { - this.concurrentLookupRequest = concurrentLookupRequest; + public void setSslProvider(String sslProvider) { + this.sslProvider = sslProvider; + } + + public String getTlsTrustStoreType() { + return this.tlsTrustStoreType; + } + + public void setTlsTrustStoreType(String tlsTrustStoreType) { + this.tlsTrustStoreType = tlsTrustStoreType; + } + + public String getTlsTrustStorePath() { + return this.tlsTrustStorePath; + } + + public void setTlsTrustStorePath(String tlsTrustStorePath) { + this.tlsTrustStorePath = tlsTrustStorePath; + } + + public String getTlsTrustStorePassword() { + return this.tlsTrustStorePassword; + } + + public void setTlsTrustStorePassword(String tlsTrustStorePassword) { + this.tlsTrustStorePassword = tlsTrustStorePassword; + } + + public String[] getTlsCiphers() { + return this.tlsCiphers; + } + + public void setTlsCiphers(String[] tlsCiphers) { + this.tlsCiphers = tlsCiphers; + } + + public String[] getTlsProtocols() { + return this.tlsProtocols; + } + + public void setTlsProtocols(String[] tlsProtocols) { + this.tlsProtocols = tlsProtocols; + } + + public long getStatsIntervalSeconds() { + return this.statsIntervalSeconds; + } + + public void setStatsIntervalSeconds(long statsIntervalSeconds) { + this.statsIntervalSeconds = statsIntervalSeconds; + } + + public int getMaxConcurrentLookupRequest() { + return this.maxConcurrentLookupRequest; + } + + public void setMaxConcurrentLookupRequest(int maxConcurrentLookupRequest) { + this.maxConcurrentLookupRequest = maxConcurrentLookupRequest; } public int getMaxLookupRequest() { @@ -732,6 +1140,14 @@ public class PulsarProperties { this.maxLookupRequest = maxLookupRequest; } + public int getMaxLookupRedirects() { + return this.maxLookupRedirects; + } + + public void setMaxLookupRedirects(int maxLookupRedirects) { + this.maxLookupRedirects = maxLookupRedirects; + } + public int getMaxNumberOfRejectedRequestPerConnection() { return this.maxNumberOfRejectedRequestPerConnection; } @@ -756,14 +1172,6 @@ public class PulsarProperties { this.connectionTimeoutMs = connectionTimeoutMs; } - public int getRequestTimeoutMs() { - return this.requestTimeoutMs; - } - - public void setRequestTimeoutMs(int requestTimeoutMs) { - this.requestTimeoutMs = requestTimeoutMs; - } - public long getInitialBackoffIntervalNanos() { return this.initialBackoffIntervalNanos; } @@ -780,31 +1188,115 @@ public class PulsarProperties { this.maxBackoffIntervalNanos = maxBackoffIntervalNanos; } + public boolean isEnableBusyWait() { + return this.enableBusyWait; + } + + public void setEnableBusyWait(boolean enableBusyWait) { + this.enableBusyWait = enableBusyWait; + } + + public long getMemoryLimitBytes() { + return this.memoryLimitBytes; + } + + public void setMemoryLimitBytes(long memoryLimitBytes) { + this.memoryLimitBytes = memoryLimitBytes; + } + + public boolean isEnableTransaction() { + return this.enableTransaction; + } + + public void setEnableTransaction(boolean enableTransaction) { + this.enableTransaction = enableTransaction; + } + + public String getDnsLookupBindAddress() { + return this.dnsLookupBindAddress; + } + + public void setDnsLookupBindAddress(String dnsLookupBindAddress) { + this.dnsLookupBindAddress = dnsLookupBindAddress; + } + + public int getDnsLookupBindPort() { + return this.dnsLookupBindPort; + } + + public void setDnsLookupBindPort(int dnsLookupBindPort) { + this.dnsLookupBindPort = dnsLookupBindPort; + } + + public String getSocks5ProxyAddress() { + return this.socks5ProxyAddress; + } + + public void setSocks5ProxyAddress(String socks5ProxyAddress) { + this.socks5ProxyAddress = socks5ProxyAddress; + } + + public String getSocks5ProxyUsername() { + return this.socks5ProxyUsername; + } + + public void setSocks5ProxyUsername(String socks5ProxyUsername) { + this.socks5ProxyUsername = socks5ProxyUsername; + } + + public String getSocks5ProxyPassword() { + return this.socks5ProxyPassword; + } + + public void setSocks5ProxyPassword(String socks5ProxyPassword) { + this.socks5ProxyPassword = socks5ProxyPassword; + } + public Map buildProperties() { PulsarProperties.Properties properties = new Properties(); PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); map.from(this::getServiceUrl).to(properties.in("serviceUrl")); + map.from(this::getListenerName).to(properties.in("listenerName")); map.from(this::getAuthPluginClassName).to(properties.in("authPluginClassName")); map.from(this::getAuthParams).to(properties.in("authParams")); + map.from(this::getAuthParamsMap).to(properties.in("authParamMap")); map.from(this::getOperationTimeoutMs).to(properties.in("operationTimeoutMs")); - map.from(this::getStatsIntervalSeconds).to(properties.in("statsIntervalSeconds")); + map.from(this::getLookupTimeoutMs).to(properties.in("lookupTimeoutMs")); map.from(this::getNumIoThreads).to(properties.in("numIoThreads")); + map.from(this::getNumListenerThreads).to(properties.in("numListenerThreads")); + map.from(this::getNumConnectionsPerBroker).to(properties.in("connectionsPerBroker")); map.from(this::isUseTcpNoDelay).to(properties.in("useTcpNoDelay")); map.from(this::isUseTls).to(properties.in("useTls")); + map.from(this::isTlsHostnameVerificationEnable).to(properties.in("tlsHostnameVerificationEnable")); map.from(this::getTlsTrustCertsFilePath).to(properties.in("tlsTrustCertsFilePath")); map.from(this::isTlsAllowInsecureConnection).to(properties.in("tlsAllowInsecureConnection")); - map.from(this::isTlsHostnameVerificationEnable).to(properties.in("tlsHostnameVerificationEnable")); - map.from(this::getConcurrentLookupRequest).to(properties.in("concurrentLookupRequest")); + map.from(this::isUseKeyStoreTls).to(properties.in("useKeyStoreTls")); + map.from(this::getSslProvider).to(properties.in("sslProvider")); + map.from(this::getTlsTrustStoreType).to(properties.in("tlsTrustStoreType")); + map.from(this::getTlsTrustStorePath).to(properties.in("tlsTrustStorePath")); + map.from(this::getTlsTrustStorePassword).to(properties.in("tlsTrustStorePassword")); + map.from(this::getTlsCiphers).to(properties.in("tlsCiphers")); + map.from(this::getTlsProtocols).to(properties.in("tlsProtocols")); + map.from(this::getStatsIntervalSeconds).to(properties.in("statsIntervalSeconds")); + map.from(this::getMaxConcurrentLookupRequest).to(properties.in("concurrentLookupRequest")); map.from(this::getMaxLookupRequest).to(properties.in("maxLookupRequest")); + map.from(this::getMaxLookupRedirects).to(properties.in("maxLookupRedirects")); map.from(this::getMaxNumberOfRejectedRequestPerConnection) .to(properties.in("maxNumberOfRejectedRequestPerConnection")); map.from(this::getKeepAliveIntervalSeconds).to(properties.in("keepAliveIntervalSeconds")); map.from(this::getConnectionTimeoutMs).to(properties.in("connectionTimeoutMs")); - map.from(this::getRequestTimeoutMs).to(properties.in("requestTimeoutMs")); map.from(this::getInitialBackoffIntervalNanos).to(properties.in("initialBackoffIntervalNanos")); map.from(this::getMaxBackoffIntervalNanos).to(properties.in("maxBackoffIntervalNanos")); + map.from(this::isEnableBusyWait).to(properties.in("enableBusyWait")); + map.from(this::getMemoryLimitBytes).to(properties.in("memoryLimitBytes")); + map.from(this::isEnableTransaction).to(properties.in("enableTransaction")); + map.from(this::getDnsLookupBindAddress).to(properties.in("dnsLookupBindAddress")); + map.from(this::getDnsLookupBindPort).to(properties.in("dnsLookupBindPort")); + map.from(this::getSocks5ProxyAddress).to(properties.in("socks5ProxyAddress")); + map.from(this::getSocks5ProxyUsername).to(properties.in("socks5ProxyUsername")); + map.from(this::getSocks5ProxyPassword).to(properties.in("socks5ProxyPassword")); return properties; } @@ -813,8 +1305,14 @@ public class PulsarProperties { public static class Listener { + /** + * AckMode for acknowledgements. Allowed values are RECORD, BATCH, MANUAL. + */ private PulsarContainerProperties.AckMode ackMode; + /** + * SchemaType of the consumed messages. + */ private SchemaType schemaType; public PulsarContainerProperties.AckMode getAckMode() { @@ -835,6 +1333,222 @@ public class PulsarProperties { } + public static class Admin { + + /** + * Pulsar service URL for the admin endpoint. + */ + private String serviceUrl; + + /** + * Class name of the clients' authentication plugin. + */ + private String authPluginClassName; + + /** + * Authentication parameter(s) of the client. + */ + private String authParams; + + /** + * Authentication parameter map of the client. + */ + private Map authParamMap; + + /** + * Path to the trusted TLS certificate file. + */ + private String tlsTrustCertsFilePath; + + /** + * Whether the client accepts untrusted TLS certificates from the broker. + */ + private boolean tlsAllowInsecureConnection = false; + + /** + * Whether the hostname is validated when the proxy creates a TLS connection with + * brokers. + */ + private boolean tlsHostnameVerificationEnable = false; + + /** + * Enable KeyStore instead of PEM type configuration if TLS is enabled. + */ + private boolean useKeyStoreTls = false; + + /** + * Name of the security provider used for SSL connections. + */ + private String sslProvider; + + /** + * File format of the trust store file. + */ + private String tlsTrustStoreType; + + /** + * Location of the trust store file. + */ + private String tlsTrustStorePath; + + /** + * Store password for the key store file. + */ + private String tlsTrustStorePassword; + + /** + * Comma-separated list of cipher suites. This is a named combination of + * authentication, encryption, MAC and key exchange algorithm used to negotiate + * the security settings for a network connection using TLS or SSL network + * protocol. By default, all the available cipher suites are supported. + */ + private String[] tlsCiphers; + + /** + * Comma-separated list of SSL protocols used to generate the SSLContext. Allowed + * values in recent JVMs are TLS, TLSv1.3, TLSv1.2 and TLSv1.1. + */ + private String[] tlsProtocols; + + public String getServiceUrl() { + return this.serviceUrl; + } + + public void setServiceUrl(String serviceUrl) { + this.serviceUrl = serviceUrl; + } + + public String getAuthPluginClassName() { + return this.authPluginClassName; + } + + public void setAuthPluginClassName(String authPluginClassName) { + this.authPluginClassName = authPluginClassName; + } + + public String getAuthParams() { + return this.authParams; + } + + public void setAuthParams(String authParams) { + this.authParams = authParams; + } + + public Map getAuthParamMap() { + return this.authParamMap; + } + + public void setAuthParamMap(Map authParamMap) { + this.authParamMap = authParamMap; + } + + public String getTlsTrustCertsFilePath() { + return this.tlsTrustCertsFilePath; + } + + public void setTlsTrustCertsFilePath(String tlsTrustCertsFilePath) { + this.tlsTrustCertsFilePath = tlsTrustCertsFilePath; + } + + public boolean isTlsAllowInsecureConnection() { + return this.tlsAllowInsecureConnection; + } + + public void setTlsAllowInsecureConnection(boolean tlsAllowInsecureConnection) { + this.tlsAllowInsecureConnection = tlsAllowInsecureConnection; + } + + public boolean isTlsHostnameVerificationEnable() { + return this.tlsHostnameVerificationEnable; + } + + public void setTlsHostnameVerificationEnable(boolean tlsHostnameVerificationEnable) { + this.tlsHostnameVerificationEnable = tlsHostnameVerificationEnable; + } + + public boolean isUseKeyStoreTls() { + return this.useKeyStoreTls; + } + + public void setUseKeyStoreTls(boolean useKeyStoreTls) { + this.useKeyStoreTls = useKeyStoreTls; + } + + public String getSslProvider() { + return this.sslProvider; + } + + public void setSslProvider(String sslProvider) { + this.sslProvider = sslProvider; + } + + public String getTlsTrustStoreType() { + return this.tlsTrustStoreType; + } + + public void setTlsTrustStoreType(String tlsTrustStoreType) { + this.tlsTrustStoreType = tlsTrustStoreType; + } + + public String getTlsTrustStorePath() { + return this.tlsTrustStorePath; + } + + public void setTlsTrustStorePath(String tlsTrustStorePath) { + this.tlsTrustStorePath = tlsTrustStorePath; + } + + public String getTlsTrustStorePassword() { + return this.tlsTrustStorePassword; + } + + public void setTlsTrustStorePassword(String tlsTrustStorePassword) { + this.tlsTrustStorePassword = tlsTrustStorePassword; + } + + public String[] getTlsCiphers() { + return this.tlsCiphers; + } + + public void setTlsCiphers(String[] tlsCiphers) { + this.tlsCiphers = tlsCiphers; + } + + public String[] getTlsProtocols() { + return this.tlsProtocols; + } + + public void setTlsProtocols(String[] tlsProtocols) { + this.tlsProtocols = tlsProtocols; + } + + public Map buildProperties() { + PulsarProperties.Properties properties = new Properties(); + + PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); + + map.from(this::getServiceUrl).to(properties.in("serviceUrl")); + map.from(this::getAuthPluginClassName).to(properties.in("authPluginClassName")); + map.from(this::getAuthParams).to(properties.in("authParams")); + map.from(this::getAuthParamMap).to(properties.in("authParamMap")); + map.from(this::getTlsTrustCertsFilePath).to(properties.in("tlsTrustCertsFilePath")); + map.from(this::isTlsAllowInsecureConnection).to(properties.in("tlsAllowInsecureConnection")); + map.from(this::isTlsHostnameVerificationEnable).to(properties.in("tlsHostnameVerificationEnable")); + map.from(this::isUseKeyStoreTls).to(properties.in("useKeyStoreTls")); + map.from(this::getSslProvider).to(properties.in("sslProvider")); + map.from(this::getTlsTrustStoreType).to(properties.in("tlsTrustStoreType")); + map.from(this::getTlsTrustStorePath).to(properties.in("tlsTrustStorePath")); + map.from(this::getTlsTrustStorePassword).to(properties.in("tlsTrustStorePassword")); + map.from(this::getTlsCiphers).to(properties.in("tlsCiphers")); + map.from(this::getTlsProtocols).to(properties.in("tlsProtocols")); + + properties.putIfAbsent("serviceUrl", "http://localhost:8080"); + + return properties; + } + + } + @SuppressWarnings("serial") private static class Properties extends HashMap { diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java index 2cc7098e..4422b257 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java @@ -43,6 +43,7 @@ import org.springframework.pulsar.config.PulsarListenerContainerFactory; import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; import org.springframework.pulsar.core.CachingPulsarProducerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; @@ -90,7 +91,7 @@ class PulsarAutoConfigurationTests { .hasSingleBean(PulsarTemplate.class).hasSingleBean(PulsarConsumerFactory.class) .hasSingleBean(ConcurrentPulsarListenerContainerFactory.class) .hasSingleBean(PulsarListenerAnnotationBeanPostProcessor.class) - .hasSingleBean(PulsarListenerEndpointRegistry.class)); + .hasSingleBean(PulsarListenerEndpointRegistry.class).hasSingleBean(PulsarAdministration.class)); } @Test @@ -161,6 +162,15 @@ class PulsarAutoConfigurationTests { .isSameAs(listenerAnnotationBeanPostProcessor)); } + @Test + void customPulsarAdministrationIsRespected() { + PulsarAdministration pulsarAdministration = mock(PulsarAdministration.class); + this.contextRunner + .withBean("customPulsarAdministration", PulsarAdministration.class, () -> pulsarAdministration) + .run((context) -> assertThat(context).hasNotFailed().getBean(PulsarAdministration.class) + .isSameAs(pulsarAdministration)); + } + @Test void customProducerInterceptorIsUsedInPulsarTemplate() { ProducerInterceptor interceptor = mock(ProducerInterceptor.class); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java index 8b688875..fa8d0e19 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java @@ -104,7 +104,8 @@ public class PulsarAdministration } private List getMatchingTopicPartitions(PulsarTopic topic, List existingTopics) { - return existingTopics.stream().filter(existing -> existing.startsWith(topic + "-partition-")).toList(); + return existingTopics.stream() + .filter(existing -> existing.startsWith(topic.getFullyQualifiedTopicName() + "-partition-")).toList(); } private void createOrModifyTopicsIfNeeded(Collection topics) { @@ -134,25 +135,26 @@ public class PulsarAdministration if (topic.isPartitioned()) { List matchingPartitions = getMatchingTopicPartitions(topic, existingTopicsInNamespace); if (matchingPartitions.isEmpty()) { - this.logger.debug(() -> "Topic " + topic + " does not exist."); + this.logger.debug(() -> "Topic " + topic.getFullyQualifiedTopicName() + " does not exist."); topicsToCreate.add(topic); } else { int numberOfExistingPartitions = matchingPartitions.size(); if (numberOfExistingPartitions < topic.numberOfPartitions()) { - this.logger.debug(() -> "Topic " + topic + " found with " + numberOfExistingPartitions - + " partitions."); + this.logger.debug(() -> "Topic " + topic.getFullyQualifiedTopicName() + " found with " + + numberOfExistingPartitions + " partitions."); topicsToModify.add(topic); } else if (numberOfExistingPartitions > topic.numberOfPartitions()) { - throw new IllegalStateException("Topic " + topic + " found with " - + numberOfExistingPartitions + " partitions. Needs to be deleted first."); + throw new IllegalStateException("Topic " + topic.getFullyQualifiedTopicName() + + " found with " + numberOfExistingPartitions + + " partitions. Needs to be deleted first."); } } } else { - if (!existingTopicsInNamespace.contains(topic.toString())) { - this.logger.debug(() -> "Topic " + topic + " does not exist."); + if (!existingTopicsInNamespace.contains(topic.getFullyQualifiedTopicName())) { + this.logger.debug(() -> "Topic " + topic.getFullyQualifiedTopicName() + " does not exist."); topicsToCreate.add(topic); } } @@ -168,8 +170,8 @@ public class PulsarAdministration } private void createTopics(PulsarAdmin admin, Set topicsToCreate) throws PulsarAdminException { - this.logger.debug(() -> "Creating topics: " - + topicsToCreate.stream().map(PulsarTopic::toString).collect(Collectors.joining(","))); + this.logger.debug(() -> "Creating topics: " + topicsToCreate.stream() + .map(PulsarTopic::getFullyQualifiedTopicName).collect(Collectors.joining(","))); for (PulsarTopic topic : topicsToCreate) { if (topic.isPartitioned()) { admin.topics().createPartitionedTopic(topic.topicName(), topic.numberOfPartitions()); @@ -181,8 +183,8 @@ public class PulsarAdministration } private void modifyTopics(PulsarAdmin admin, Set topicsToModify) throws PulsarAdminException { - this.logger.debug(() -> "Modifying topics: " - + topicsToModify.stream().map(PulsarTopic::toString).collect(Collectors.joining(","))); + this.logger.debug(() -> "Modifying topics: " + topicsToModify.stream() + .map(PulsarTopic::getFullyQualifiedTopicName).collect(Collectors.joining(","))); for (PulsarTopic topic : topicsToModify) { admin.topics().updatePartitionedTopic(topic.topicName(), topic.numberOfPartitions()); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTopic.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTopic.java index f9978310..895dec4c 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTopic.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTopic.java @@ -69,8 +69,7 @@ public record PulsarTopic(String topicName, int numberOfPartitions) { * Get the fully-qualified name of the topic. * @return the fully-qualified topic name */ - @Override - public String toString() { + public String getFullyQualifiedTopicName() { TopicComponents components = this.getComponents(); return components.domain + "://" + components.tenant + "/" + components.namespace + "/" + components.name; } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java index d48d3484..97706264 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java @@ -48,7 +48,7 @@ public class PulsarContainerProperties { */ BATCH, /** - * Recod ack mode. + * Record ack mode. */ RECORD, /** diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationTests.java index c946024f..519c89c1 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationTests.java @@ -57,11 +57,11 @@ public class PulsarAdministrationTests extends AbstractContainerBaseTests { List expectedTopics = expected.stream().mapMulti((topic, consumer) -> { if (topic.isPartitioned()) { for (int i = 0; i < topic.numberOfPartitions(); i++) { - consumer.accept(topic + "-partition-" + i); + consumer.accept(topic.getFullyQualifiedTopicName() + "-partition-" + i); } } else { - consumer.accept(topic.toString()); + consumer.accept(topic.getFullyQualifiedTopicName()); } }).toList();