Spring/spring-boot
Spring/spring-boot
1
0
Fork 0
mirror of https://github.com/spring-projects/spring-boot.git synced 2024-07-15 01:07:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add support for Pulsar cluster-level failover

Browse Source 
See gh-38559
This commit is contained in:
Swamy Mavuri 2023-11-26 01:58:39 +05:30 committed by Moritz Halbritter
parent 4b157ceaf2
commit c3e3372336
6 changed files with 333 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

111
spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarProperties.java
View File

@ -19,10 +19,12 @@ package org.springframework.boot.autoconfigure.pulsar;
import java.time.Duration;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;
import org.apache.pulsar.client.api.AutoClusterFailoverBuilder.FailoverPolicy;
import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.HashingScheme;
import org.apache.pulsar.client.api.MessageRoutingMode;
@ -42,6 +44,7 @@ import org.springframework.util.Assert;
*
* @author Chris Bono
* @author Phillip Webb
* @author Swamy Mavuri
* @since 3.2.0
*/
@ConfigurationProperties("spring.pulsar")
@ -128,6 +131,11 @@ public class PulsarProperties {
*/
private final Authentication authentication = new Authentication();
/**
* Failover settings.
*/
private final Failover failover = new Failover();
public String getServiceUrl() {
return this.serviceUrl;
}
@ -164,6 +172,10 @@ public class PulsarProperties {
return this.authentication;
}
public Failover getFailover() {
return this.failover;
}
}
public static class Admin {
@ -887,4 +899,103 @@ public class PulsarProperties {
}
public static class Failover {
/**
* Cluster Failover Policy.
*/
private FailoverPolicy failoverPolicy = FailoverPolicy.ORDER;
/**
* Delay before the Pulsar client switches from the primary cluster to the backup
* cluster.
*/
private Duration failOverDelay;
/**
* Delay before the Pulsar client switches from the backup cluster to the primary
* cluster.
*/
private Duration switchBackDelay;
/**
* Frequency of performing a probe task.
*/
private Duration checkInterval;
/**
* List of backupClusters The backup cluster is chosen in the sequence of the
* given list. If all backup clusters are available, the Pulsar client chooses the
* first backup cluster.
*/
private List<BackupCluster> backupClusters = new LinkedList<>();
public FailoverPolicy getFailoverPolicy() {
return this.failoverPolicy;
}
public void setFailoverPolicy(FailoverPolicy failoverPolicy) {
this.failoverPolicy = failoverPolicy;
}
public Duration getFailOverDelay() {
return this.failOverDelay;
}
public void setFailOverDelay(Duration failOverDelay) {
this.failOverDelay = failOverDelay;
}
public Duration getSwitchBackDelay() {
return this.switchBackDelay;
}
public void setSwitchBackDelay(Duration switchBackDelay) {
this.switchBackDelay = switchBackDelay;
}
public Duration getCheckInterval() {
return this.checkInterval;
}
public void setCheckInterval(Duration checkInterval) {
this.checkInterval = checkInterval;
}
public List<BackupCluster> getBackupClusters() {
return this.backupClusters;
}
public void setBackupClusters(List<BackupCluster> backupClusters) {
this.backupClusters = backupClusters;
}
public static class BackupCluster {
/**
* Pulsar service URL in the format '(pulsar|pulsar+ssl)://host:port'.
*/
private String serviceUrl = "pulsar://localhost:6650";
/**
* Authentication settings.
*/
private final Authentication authentication = new Authentication();
public String getServiceUrl() {
return this.serviceUrl;
}
public void setServiceUrl(String serviceUrl) {
this.serviceUrl = serviceUrl;
}
public Authentication getAuthentication() {
return this.authentication;
}
}
}
}

48
spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarPropertiesMapper.java
View File

@ -18,6 +18,7 @@ package org.springframework.boot.autoconfigure.pulsar;
import java.time.Duration;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
@ -25,11 +26,16 @@ import java.util.function.BiConsumer;
import java.util.function.Consumer;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.AutoClusterFailoverBuilder;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
import org.apache.pulsar.client.api.ReaderBuilder;
import org.apache.pulsar.client.api.ServiceUrlProvider;
import org.apache.pulsar.client.impl.AutoClusterFailover.AutoClusterFailoverBuilderImpl;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.springframework.boot.context.properties.PropertyMapper;
@ -42,6 +48,7 @@ import org.springframework.util.StringUtils;
*
* @author Chris Bono
* @author Phillip Webb
* @author Swamy Mavuri
*/
final class PulsarPropertiesMapper {
@ -54,11 +61,50 @@ final class PulsarPropertiesMapper {
void customizeClientBuilder(ClientBuilder clientBuilder, PulsarConnectionDetails connectionDetails) {
PulsarProperties.Client properties = this.properties.getClient();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(connectionDetails::getBrokerUrl).to(clientBuilder::serviceUrl);
map.from(properties::getConnectionTimeout).to(timeoutProperty(clientBuilder::connectionTimeout));
map.from(properties::getOperationTimeout).to(timeoutProperty(clientBuilder::operationTimeout));
map.from(properties::getLookupTimeout).to(timeoutProperty(clientBuilder::lookupTimeout));
customizeAuthentication(clientBuilder::authentication, properties.getAuthentication());
customizeServiceUrlProviderBuilder(clientBuilder::serviceUrl, clientBuilder::serviceUrlProvider, properties,
connectionDetails);
}
private void customizeServiceUrlProviderBuilder(Consumer<String> serviceUrlConsumer,
Consumer<ServiceUrlProvider> serviceUrlProviderConsumer, PulsarProperties.Client properties,
PulsarConnectionDetails connectionDetails) {
PulsarProperties.Failover failoverProperties = properties.getFailover();
if (!failoverProperties.getBackupClusters().isEmpty()) {
Map<String, Authentication> secondaryAuths = new LinkedHashMap<>();
failoverProperties.getBackupClusters().forEach((cluster) -> {
PulsarProperties.Authentication authentication = cluster.getAuthentication();
if (authentication.getPluginClassName() != null) {
customizeAuthentication((authPluginClassName, authParams) -> secondaryAuths
.put(cluster.getServiceUrl(), AuthenticationFactory.create(authPluginClassName, authParams)),
authentication);
}
else {
secondaryAuths.put(cluster.getServiceUrl(), null);
}
});
AutoClusterFailoverBuilder autoClusterFailoverBuilder = new AutoClusterFailoverBuilderImpl();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(connectionDetails::getBrokerUrl).to(autoClusterFailoverBuilder::primary);
map.from(new ArrayList<>(secondaryAuths.keySet())).to(autoClusterFailoverBuilder::secondary);
map.from(failoverProperties::getFailoverPolicy).to(autoClusterFailoverBuilder::failoverPolicy);
map.from(failoverProperties::getFailOverDelay)
.to(timeoutProperty(autoClusterFailoverBuilder::failoverDelay));
map.from(failoverProperties::getSwitchBackDelay)
.to(timeoutProperty(autoClusterFailoverBuilder::switchBackDelay));
map.from(failoverProperties::getCheckInterval)
.to(timeoutProperty(autoClusterFailoverBuilder::checkInterval));
map.from(secondaryAuths).to(autoClusterFailoverBuilder::secondaryAuthentication);
serviceUrlProviderConsumer.accept(autoClusterFailoverBuilder.build());
}
else {
serviceUrlConsumer.accept(connectionDetails.getBrokerUrl());
}
}
void customizeAdminBuilder(PulsarAdminBuilder adminBuilder, PulsarConnectionDetails connectionDetails) {

63
spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/pulsar/MockAuthentication.java Normal file
View File

@ -0,0 +1,63 @@
/*
* 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 org.springframework.boot.autoconfigure.pulsar;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
import org.apache.pulsar.client.api.PulsarClientException;
/**
* Test plugin-class-name for Authentication
*
* @author Swamy Mavuri
*/
public class MockAuthentication implements Authentication {
public Map<String, String> authParamsMap = new HashMap<>();
@Override
public String getAuthMethodName() {
return null;
}
@Override
public AuthenticationDataProvider getAuthData() {
return null;
}
@Override
@Deprecated
public void configure(Map<String, String> authParams) {
this.authParamsMap = authParams;
}
@Override
public void start() throws PulsarClientException {
}
@Override
public void close() throws IOException {
}
}

44
spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/pulsar/PulsarConfigurationTests.java
View File

@ -16,21 +16,26 @@
package org.springframework.boot.autoconfigure.pulsar;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.AutoClusterFailover;
import org.apache.pulsar.common.schema.KeyValueEncodingType;
import org.assertj.core.api.InstanceOfAssertFactories;
import org.assertj.core.api.InstanceOfAssertFactory;
import org.assertj.core.api.MapAssert;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.mockito.InOrder;
import org.mockito.Mockito;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.TestConfiguration;
@ -48,10 +53,12 @@ import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.core.SchemaResolver.SchemaResolverCustomizer;
import org.springframework.pulsar.core.TopicResolver;
import org.springframework.pulsar.function.PulsarFunctionAdministration;
import org.springframework.test.util.ReflectionTestUtils;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.entry;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
/**
@ -61,6 +68,7 @@ import static org.mockito.Mockito.mock;
* @author Alexander Preuß
* @author Soby Chacko
* @author Phillip Webb
* @author Swamy Mavuri
*/
class PulsarConfigurationTests {
@ -113,6 +121,42 @@ class PulsarConfigurationTests {
});
}
@Test
void whenHasUserDefinedFailoverPropertiesAddsToClient() {
PulsarConnectionDetails connectionDetails = mock(PulsarConnectionDetails.class);
given(connectionDetails.getBrokerUrl()).willReturn("connectiondetails");
PulsarConfigurationTests.this.contextRunner.withBean(PulsarConnectionDetails.class, () -> connectionDetails)
.withPropertyValues("spring.pulsar.client.service-url=properties",
"spring.pulsar.client.failover.backup-clusters[0].service-url=backup-cluster-1",
"spring.pulsar.client.failover.failover-delay=15s",
"spring.pulsar.client.failover.switch-back-delay=30s",
"spring.pulsar.client.failover.check-interval=5s",
"spring.pulsar.client.failover.backup-clusters[1].service-url=backup-cluster-2",
"spring.pulsar.client.failover.backup-clusters[1].authentication.plugin-class-name=org.springframework.boot.autoconfigure.pulsar.MockAuthentication",
"spring.pulsar.client.failover.backup-clusters[1].authentication.param.token=1234")
.run((context) -> {
DefaultPulsarClientFactory clientFactory = context.getBean(DefaultPulsarClientFactory.class);
PulsarProperties pulsarProperties = context.getBean(PulsarProperties.class);
ClientBuilder target = mock(ClientBuilder.class);
BiConsumer<PulsarClientBuilderCustomizer, ClientBuilder> customizeAction = PulsarClientBuilderCustomizer::customize;
PulsarClientBuilderCustomizer pulsarClientBuilderCustomizer = (PulsarClientBuilderCustomizer) ReflectionTestUtils
.getField(clientFactory, "customizer");
customizeAction.accept(pulsarClientBuilderCustomizer, target);
InOrder ordered = inOrder(target);
ordered.verify(target).serviceUrlProvider(Mockito.any(AutoClusterFailover.class));
assertThat(pulsarProperties.getClient().getFailover().getFailOverDelay())
.isEqualTo(Duration.ofSeconds(15));
assertThat(pulsarProperties.getClient().getFailover().getSwitchBackDelay())
.isEqualTo(Duration.ofSeconds(30));
assertThat(pulsarProperties.getClient().getFailover().getCheckInterval())
.isEqualTo(Duration.ofSeconds(5));
assertThat(pulsarProperties.getClient().getFailover().getBackupClusters().size()).isEqualTo(2);
});
}
@TestConfiguration(proxyBeanMethods = false)
static class PulsarClientBuilderCustomizersConfig {

34
spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/pulsar/PulsarPropertiesMapperTests.java
View File

@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.api.AutoClusterFailoverBuilder.FailoverPolicy;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.ConsumerBuilder;
@ -34,10 +35,13 @@ import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
import org.apache.pulsar.client.api.ReaderBuilder;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.AutoClusterFailover;
import org.apache.pulsar.common.schema.SchemaType;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Consumer;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Failover.BackupCluster;
import org.springframework.pulsar.listener.PulsarContainerProperties;
import static org.assertj.core.api.Assertions.assertThat;
@ -50,6 +54,7 @@ import static org.mockito.Mockito.mock;
*
* @author Chris Bono
* @author Phillip Webb
* @author Swamy Mavuri
*/
class PulsarPropertiesMapperTests {
@ -93,6 +98,35 @@ class PulsarPropertiesMapperTests {
then(builder).should().serviceUrl("https://used.example.com");
}
@Test
void customizeClientBuilderWhenHasFailover() {
BackupCluster backupCluster1 = new BackupCluster();
backupCluster1.setServiceUrl("backup-cluster-1");
Map<String, String> params = Map.of("param", "name");
backupCluster1.getAuthentication()
.setPluginClassName("org.springframework.boot.autoconfigure.pulsar.MockAuthentication");
backupCluster1.getAuthentication().setParam(params);
BackupCluster backupCluster2 = new BackupCluster();
backupCluster2.setServiceUrl("backup-cluster-2");
PulsarProperties properties = new PulsarProperties();
properties.getClient().setServiceUrl("https://used.example.com");
properties.getClient().getFailover().setFailoverPolicy(FailoverPolicy.ORDER);
properties.getClient().getFailover().setCheckInterval(Duration.ofSeconds(5));
properties.getClient().getFailover().setFailOverDelay(Duration.ofSeconds(30));
properties.getClient().getFailover().setSwitchBackDelay(Duration.ofSeconds(30));
properties.getClient().getFailover().setBackupClusters(List.of(backupCluster1, backupCluster2));
PulsarConnectionDetails connectionDetails = mock(PulsarConnectionDetails.class);
given(connectionDetails.getBrokerUrl()).willReturn("https://used.example.com");
ClientBuilder builder = mock(ClientBuilder.class);
new PulsarPropertiesMapper(properties).customizeClientBuilder(builder,
new PropertiesPulsarConnectionDetails(properties));
then(builder).should().serviceUrlProvider(Mockito.any(AutoClusterFailover.class));
}
@Test
void customizeAdminBuilderWhenHasNoAuthentication() {
PulsarProperties properties = new PulsarProperties();

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

@ -18,6 +18,7 @@ package org.springframework.boot.autoconfigure.pulsar;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.pulsar.client.api.CompressionType;
@ -34,6 +35,8 @@ import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Defaults.SchemaInfo;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Defaults.TypeMapping;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Failover;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Failover.BackupCluster;
import org.springframework.boot.context.properties.bind.BindException;
import org.springframework.boot.context.properties.bind.Binder;
import org.springframework.boot.context.properties.source.MapConfigurationPropertySource;
@ -48,6 +51,7 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
* @author Christophe Bornet
* @author Soby Chacko
* @author Phillip Webb
* @author Swamy Mavuri
*/
class PulsarPropertiesTests {
@ -82,6 +86,36 @@ class PulsarPropertiesTests {
assertThat(properties.getAuthentication().getParam()).containsEntry("token", "1234");
}
@Test
void bindFailover() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.client.service-url", "my-service-url");
map.put("spring.pulsar.client.failover.failover-delay", "30s");
map.put("spring.pulsar.client.failover.switch-back-delay", "15s");
map.put("spring.pulsar.client.failover.check-interval", "1s");
map.put("spring.pulsar.client.failover.backup-clusters[0].service-url", "backup-service-url-1");
map.put("spring.pulsar.client.failover.backup-clusters[0].authentication.plugin-class-name",
"com.example.MyAuth1");
map.put("spring.pulsar.client.failover.backup-clusters[0].authentication.param.token", "1234");
map.put("spring.pulsar.client.failover.backup-clusters[1].service-url", "backup-service-url-2");
map.put("spring.pulsar.client.failover.backup-clusters[1].authentication.plugin-class-name",
"com.example.MyAuth2");
map.put("spring.pulsar.client.failover.backup-clusters[1].authentication.param.token", "5678");
PulsarProperties.Client properties = bindPropeties(map).getClient();
Failover failoverProperties = properties.getFailover();
List<BackupCluster> backupClusters = properties.getFailover().getBackupClusters();
assertThat(properties.getServiceUrl()).isEqualTo("my-service-url");
assertThat(failoverProperties.getFailOverDelay()).isEqualTo(Duration.ofMillis(30000));
assertThat(failoverProperties.getSwitchBackDelay()).isEqualTo(Duration.ofMillis(15000));
assertThat(failoverProperties.getCheckInterval()).isEqualTo(Duration.ofMillis(1000));
assertThat(backupClusters.get(0).getServiceUrl()).isEqualTo("backup-service-url-1");
assertThat(backupClusters.get(0).getAuthentication().getPluginClassName()).isEqualTo("com.example.MyAuth1");
assertThat(backupClusters.get(0).getAuthentication().getParam()).containsEntry("token", "1234");
assertThat(backupClusters.get(1).getServiceUrl()).isEqualTo("backup-service-url-2");
assertThat(backupClusters.get(1).getAuthentication().getPluginClassName()).isEqualTo("com.example.MyAuth2");
assertThat(backupClusters.get(1).getAuthentication().getParam()).containsEntry("token", "5678");
}
}
@Nested