mirror of
https://github.com/spring-projects/spring-boot.git
synced 2024-08-29 03:06:45 +08:00
Polish "Add support for Pulsar cluster-level failover"
See gh-38559
This commit is contained in:
parent
c3e3372336
commit
f696190d83
@ -19,7 +19,6 @@ package org.springframework.boot.autoconfigure.pulsar;
|
|||||||
import java.time.Duration;
|
import java.time.Duration;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.LinkedHashMap;
|
import java.util.LinkedHashMap;
|
||||||
import java.util.LinkedList;
|
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import java.util.regex.Pattern;
|
import java.util.regex.Pattern;
|
||||||
@ -928,7 +927,7 @@ public class PulsarProperties {
|
|||||||
* given list. If all backup clusters are available, the Pulsar client chooses the
|
* given list. If all backup clusters are available, the Pulsar client chooses the
|
||||||
* first backup cluster.
|
* first backup cluster.
|
||||||
*/
|
*/
|
||||||
private List<BackupCluster> backupClusters = new LinkedList<>();
|
private List<BackupCluster> backupClusters = new ArrayList<>();
|
||||||
|
|
||||||
public FailoverPolicy getFailoverPolicy() {
|
public FailoverPolicy getFailoverPolicy() {
|
||||||
return this.failoverPolicy;
|
return this.failoverPolicy;
|
||||||
|
@ -73,38 +73,32 @@ final class PulsarPropertiesMapper {
|
|||||||
Consumer<ServiceUrlProvider> serviceUrlProviderConsumer, PulsarProperties.Client properties,
|
Consumer<ServiceUrlProvider> serviceUrlProviderConsumer, PulsarProperties.Client properties,
|
||||||
PulsarConnectionDetails connectionDetails) {
|
PulsarConnectionDetails connectionDetails) {
|
||||||
PulsarProperties.Failover failoverProperties = properties.getFailover();
|
PulsarProperties.Failover failoverProperties = properties.getFailover();
|
||||||
if (!failoverProperties.getBackupClusters().isEmpty()) {
|
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());
|
serviceUrlConsumer.accept(connectionDetails.getBrokerUrl());
|
||||||
|
return;
|
||||||
}
|
}
|
||||||
|
Map<String, Authentication> secondaryAuths = new LinkedHashMap<>();
|
||||||
|
failoverProperties.getBackupClusters().forEach((cluster) -> {
|
||||||
|
PulsarProperties.Authentication authentication = cluster.getAuthentication();
|
||||||
|
if (authentication.getPluginClassName() == null) {
|
||||||
|
secondaryAuths.put(cluster.getServiceUrl(), null);
|
||||||
|
}
|
||||||
|
else {
|
||||||
|
customizeAuthentication((authPluginClassName, authParams) -> secondaryAuths.put(cluster.getServiceUrl(),
|
||||||
|
AuthenticationFactory.create(authPluginClassName, authParams)), authentication);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
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());
|
||||||
}
|
}
|
||||||
|
|
||||||
void customizeAdminBuilder(PulsarAdminBuilder adminBuilder, PulsarConnectionDetails connectionDetails) {
|
void customizeAdminBuilder(PulsarAdminBuilder adminBuilder, PulsarConnectionDetails connectionDetails) {
|
||||||
|
@ -29,7 +29,7 @@ import org.apache.pulsar.client.api.PulsarClientException;
|
|||||||
*
|
*
|
||||||
* @author Swamy Mavuri
|
* @author Swamy Mavuri
|
||||||
*/
|
*/
|
||||||
|
@SuppressWarnings("deprecation")
|
||||||
public class MockAuthentication implements Authentication {
|
public class MockAuthentication implements Authentication {
|
||||||
|
|
||||||
public Map<String, String> authParamsMap = new HashMap<>();
|
public Map<String, String> authParamsMap = new HashMap<>();
|
||||||
@ -45,7 +45,6 @@ public class MockAuthentication implements Authentication {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@Deprecated
|
|
||||||
public void configure(Map<String, String> authParams) {
|
public void configure(Map<String, String> authParams) {
|
||||||
this.authParamsMap = authParams;
|
this.authParamsMap = authParams;
|
||||||
}
|
}
|
||||||
|
@ -125,7 +125,6 @@ class PulsarConfigurationTests {
|
|||||||
void whenHasUserDefinedFailoverPropertiesAddsToClient() {
|
void whenHasUserDefinedFailoverPropertiesAddsToClient() {
|
||||||
PulsarConnectionDetails connectionDetails = mock(PulsarConnectionDetails.class);
|
PulsarConnectionDetails connectionDetails = mock(PulsarConnectionDetails.class);
|
||||||
given(connectionDetails.getBrokerUrl()).willReturn("connectiondetails");
|
given(connectionDetails.getBrokerUrl()).willReturn("connectiondetails");
|
||||||
|
|
||||||
PulsarConfigurationTests.this.contextRunner.withBean(PulsarConnectionDetails.class, () -> connectionDetails)
|
PulsarConfigurationTests.this.contextRunner.withBean(PulsarConnectionDetails.class, () -> connectionDetails)
|
||||||
.withPropertyValues("spring.pulsar.client.service-url=properties",
|
.withPropertyValues("spring.pulsar.client.service-url=properties",
|
||||||
"spring.pulsar.client.failover.backup-clusters[0].service-url=backup-cluster-1",
|
"spring.pulsar.client.failover.backup-clusters[0].service-url=backup-cluster-1",
|
||||||
@ -138,7 +137,6 @@ class PulsarConfigurationTests {
|
|||||||
.run((context) -> {
|
.run((context) -> {
|
||||||
DefaultPulsarClientFactory clientFactory = context.getBean(DefaultPulsarClientFactory.class);
|
DefaultPulsarClientFactory clientFactory = context.getBean(DefaultPulsarClientFactory.class);
|
||||||
PulsarProperties pulsarProperties = context.getBean(PulsarProperties.class);
|
PulsarProperties pulsarProperties = context.getBean(PulsarProperties.class);
|
||||||
|
|
||||||
ClientBuilder target = mock(ClientBuilder.class);
|
ClientBuilder target = mock(ClientBuilder.class);
|
||||||
BiConsumer<PulsarClientBuilderCustomizer, ClientBuilder> customizeAction = PulsarClientBuilderCustomizer::customize;
|
BiConsumer<PulsarClientBuilderCustomizer, ClientBuilder> customizeAction = PulsarClientBuilderCustomizer::customize;
|
||||||
PulsarClientBuilderCustomizer pulsarClientBuilderCustomizer = (PulsarClientBuilderCustomizer) ReflectionTestUtils
|
PulsarClientBuilderCustomizer pulsarClientBuilderCustomizer = (PulsarClientBuilderCustomizer) ReflectionTestUtils
|
||||||
@ -146,7 +144,6 @@ class PulsarConfigurationTests {
|
|||||||
customizeAction.accept(pulsarClientBuilderCustomizer, target);
|
customizeAction.accept(pulsarClientBuilderCustomizer, target);
|
||||||
InOrder ordered = inOrder(target);
|
InOrder ordered = inOrder(target);
|
||||||
ordered.verify(target).serviceUrlProvider(Mockito.any(AutoClusterFailover.class));
|
ordered.verify(target).serviceUrlProvider(Mockito.any(AutoClusterFailover.class));
|
||||||
|
|
||||||
assertThat(pulsarProperties.getClient().getFailover().getFailOverDelay())
|
assertThat(pulsarProperties.getClient().getFailover().getFailOverDelay())
|
||||||
.isEqualTo(Duration.ofSeconds(15));
|
.isEqualTo(Duration.ofSeconds(15));
|
||||||
assertThat(pulsarProperties.getClient().getFailover().getSwitchBackDelay())
|
assertThat(pulsarProperties.getClient().getFailover().getSwitchBackDelay())
|
||||||
|
@ -106,10 +106,8 @@ class PulsarPropertiesMapperTests {
|
|||||||
backupCluster1.getAuthentication()
|
backupCluster1.getAuthentication()
|
||||||
.setPluginClassName("org.springframework.boot.autoconfigure.pulsar.MockAuthentication");
|
.setPluginClassName("org.springframework.boot.autoconfigure.pulsar.MockAuthentication");
|
||||||
backupCluster1.getAuthentication().setParam(params);
|
backupCluster1.getAuthentication().setParam(params);
|
||||||
|
|
||||||
BackupCluster backupCluster2 = new BackupCluster();
|
BackupCluster backupCluster2 = new BackupCluster();
|
||||||
backupCluster2.setServiceUrl("backup-cluster-2");
|
backupCluster2.setServiceUrl("backup-cluster-2");
|
||||||
|
|
||||||
PulsarProperties properties = new PulsarProperties();
|
PulsarProperties properties = new PulsarProperties();
|
||||||
properties.getClient().setServiceUrl("https://used.example.com");
|
properties.getClient().setServiceUrl("https://used.example.com");
|
||||||
properties.getClient().getFailover().setFailoverPolicy(FailoverPolicy.ORDER);
|
properties.getClient().getFailover().setFailoverPolicy(FailoverPolicy.ORDER);
|
||||||
@ -117,10 +115,8 @@ class PulsarPropertiesMapperTests {
|
|||||||
properties.getClient().getFailover().setFailOverDelay(Duration.ofSeconds(30));
|
properties.getClient().getFailover().setFailOverDelay(Duration.ofSeconds(30));
|
||||||
properties.getClient().getFailover().setSwitchBackDelay(Duration.ofSeconds(30));
|
properties.getClient().getFailover().setSwitchBackDelay(Duration.ofSeconds(30));
|
||||||
properties.getClient().getFailover().setBackupClusters(List.of(backupCluster1, backupCluster2));
|
properties.getClient().getFailover().setBackupClusters(List.of(backupCluster1, backupCluster2));
|
||||||
|
|
||||||
PulsarConnectionDetails connectionDetails = mock(PulsarConnectionDetails.class);
|
PulsarConnectionDetails connectionDetails = mock(PulsarConnectionDetails.class);
|
||||||
given(connectionDetails.getBrokerUrl()).willReturn("https://used.example.com");
|
given(connectionDetails.getBrokerUrl()).willReturn("https://used.example.com");
|
||||||
|
|
||||||
ClientBuilder builder = mock(ClientBuilder.class);
|
ClientBuilder builder = mock(ClientBuilder.class);
|
||||||
new PulsarPropertiesMapper(properties).customizeClientBuilder(builder,
|
new PulsarPropertiesMapper(properties).customizeClientBuilder(builder,
|
||||||
new PropertiesPulsarConnectionDetails(properties));
|
new PropertiesPulsarConnectionDetails(properties));
|
||||||
|
Loading…
Reference in New Issue
Block a user