Merge pull request #38559 from swamymavuri

* pr/38559:
  Polish "Add support for Pulsar cluster-level failover"
  Add support for Pulsar cluster-level failover

Closes gh-38559

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

@@ -23,6 +23,7 @@ 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 +43,7 @@ import org.springframework.util.Assert;
  *
  * @author Chris Bono
  * @author Phillip Webb
+ * @author Swamy Mavuri
  * @since 3.2.0
  */
 @ConfigurationProperties("spring.pulsar")
@@ -128,6 +130,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 +171,10 @@ public class PulsarProperties {
 			return this.authentication;
 		}
 
+		public Failover getFailover() {
+			return this.failover;
+		}
+
 	}
 
 	public static class Admin {
@@ -887,4 +898,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 ArrayList<>();
+
+		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;
+			}
+
+		}
+
+	}
+
 }

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

@@ -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,44 @@ 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()) {
+			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) {

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/pulsar/MockAuthentication.java

@@ -0,0 +1,62 @@
+/*
+ * 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
+ */
+@SuppressWarnings("deprecation")
+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
+	public void configure(Map<String, String> authParams) {
+		this.authParamsMap = authParams;
+	}
+
+	@Override
+	public void start() throws PulsarClientException {
+
+	}
+
+	@Override
+	public void close() throws IOException {
+
+	}
+
+}

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

@@ -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,39 @@ 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 {
 

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

@@ -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,31 @@ 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();

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

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