GH-1325: Queue API Deprecations and Improvements

Resolves #1325

Author: garyrussell

spring-amqp/src/main/java/org/springframework/amqp/core/AnonymousQueue.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -56,15 +56,16 @@ public AnonymousQueue(org.springframework.amqp.core.NamingStrategy namingStrateg
 
 	/**
 	 * Construct a queue with a name provided by the supplied naming strategy with the
-	 * supplied arguments.
+	 * supplied arguments. By default, these queues will be created on the node the
+	 * application is connected to.
 	 * @param namingStrategy the naming strategy.
 	 * @param arguments the arguments.
 	 * @since 2.1
 	 */
 	public AnonymousQueue(org.springframework.amqp.core.NamingStrategy namingStrategy, Map<String, Object> arguments) {
 		super(namingStrategy.generateName(), false, true, true, arguments);
-		if (!getArguments().containsKey(X_QUEUE_MASTER_LOCATOR)) {
-			setMasterLocator("client-local");
+		if (!getArguments().containsKey(X_QUEUE_LEADER_LOCATOR)) {
+			setLeaderLocator("client-local");
 		}
 	}
 

spring-amqp/src/main/java/org/springframework/amqp/core/Queue.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -35,9 +35,17 @@ public class Queue extends AbstractDeclarable implements Cloneable {
 	/**
 	 * Argument key for the master locator.
 	 * @since 2.1
+	 * @deprecated in favor of {@link #X_QUEUE_LEADER_LOCATOR}.
 	 */
+	@Deprecated
 	public static final String X_QUEUE_MASTER_LOCATOR = "x-queue-master-locator";
 
+	/**
+	 * Argument key for the queue leader locator.
+	 * @since 2.1
+	 */
+	public static final String X_QUEUE_LEADER_LOCATOR = "x-queue-master-locator";
+
 	private final String name;
 
 	private final boolean durable;
@@ -161,13 +169,29 @@ public String getActualName() {
 	 * Set the master locator strategy argument for this queue.
 	 * @param locator the locator; null to clear the argument.
 	 * @since 2.1
+	 * @deprecated in favor of {@link #setLeaderLocator(String)}.
 	 */
+	@Deprecated
 	public final void setMasterLocator(@Nullable String locator) {
 		if (locator == null) {
-			removeArgument(X_QUEUE_MASTER_LOCATOR);
+			removeArgument(X_QUEUE_LEADER_LOCATOR);
+		}
+		else {
+			addArgument(X_QUEUE_LEADER_LOCATOR, locator);
+		}
+	}
+
+	/**
+	 * Set the leader locator strategy argument for this queue.
+	 * @param locator the locator; null to clear the argument.
+	 * @since 2.3.7
+	 */
+	public final void setLeaderLocator(@Nullable String locator) {
+		if (locator == null) {
+			removeArgument(X_QUEUE_LEADER_LOCATOR);
 		}
 		else {
-			addArgument(X_QUEUE_MASTER_LOCATOR, locator);
+			addArgument(X_QUEUE_LEADER_LOCATOR, locator);
 		}
 	}
 

spring-amqp/src/main/java/org/springframework/amqp/core/QueueBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -22,6 +22,8 @@
  * Builds a Spring AMQP Queue using a fluent API.
  *
  * @author Maciej Walkowiak
+ * @author Gary Russell
+ *
  * @since 1.6
  *
  */
@@ -228,11 +230,25 @@ public QueueBuilder lazy() {
 	 * or {@link MasterLocator#random}.
 	 * @return the builder.
 	 * @since 2.2
+	 * @deprecated in favor of {@link #leaderLocator(LeaderLocator)}.
 	 */
+	@Deprecated
 	public QueueBuilder masterLocator(MasterLocator locator) {
 		return withArgument("x-queue-master-locator", locator.getValue());
 	}
 
+	/**
+	 * Set the master locator mode which determines which node a queue master will be
+	 * located on a cluster of nodes.
+	 * @param locator {@link MasterLocator#minMasters}, {@link MasterLocator#clientLocal}
+	 * or {@link MasterLocator#random}.
+	 * @return the builder.
+	 * @since 2.2
+	 */
+	public QueueBuilder leaderLocator(LeaderLocator locator) {
+		return withArgument("x-queue-master-locator", locator.getValue());
+	}
+
 	/**
 	 * Set the 'x-single-active-consumer' queue argument.
 	 * @return the builder.
@@ -270,6 +286,9 @@ public Queue build() {
 		return new Queue(this.name, this.durable, this.exclusive, this.autoDelete, getArguments());
 	}
 
+	/**
+	 * Overflow argument values.
+	 */
 	public enum Overflow {
 
 		/**
@@ -288,12 +307,20 @@ public enum Overflow {
 			this.value = value;
 		}
 
+		/**
+		 * Return the value.
+		 * @return the value.
+		 */
 		public String getValue() {
 			return this.value;
 		}
 
 	}
 
+	/**
+	 * @deprecated in favor of {@link LeaderLocator}.
+	 */
+	@Deprecated
 	public enum MasterLocator {
 
 		/**
@@ -317,6 +344,49 @@ public enum MasterLocator {
 			this.value = value;
 		}
 
+		/**
+		 * Return the value.
+		 * @return the value.
+		 */
+		public String getValue() {
+			return this.value;
+		}
+
+	}
+
+	/**
+	 * Locate the queue leader.
+	 *
+	 * @since 2.3.7
+	 *
+	 */
+	public enum LeaderLocator {
+
+		/**
+		 * Deploy on the node with the fewest queue leaders.
+		 */
+		minLeaders("min-masters"),
+
+		/**
+		 * Deploy on the node we are connected to.
+		 */
+		clientLocal("client-local"),
+
+		/**
+		 * Deploy on a random node.
+		 */
+		random("random");
+
+		private final String value;
+
+		LeaderLocator(String value) {
+			this.value = value;
+		}
+
+		/**
+		 * Return the value.
+		 * @return the value.
+		 */
 		public String getValue() {
 			return this.value;
 		}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2020 the original author or authors.
+ * Copyright 2014-2021 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.
@@ -35,7 +35,7 @@
 import org.springframework.amqp.core.ExchangeBuilder;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.core.QueueBuilder;
-import org.springframework.amqp.core.QueueBuilder.MasterLocator;
+import org.springframework.amqp.core.QueueBuilder.LeaderLocator;
 import org.springframework.amqp.core.QueueBuilder.Overflow;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -111,7 +111,7 @@ void testQueueArgs1() throws MalformedURLException, URISyntaxException, Interrup
 		assertThat(arguments.get("x-dead-letter-routing-key")).isEqualTo("reply.dlrk");
 		assertThat(arguments.get("x-max-priority")).isEqualTo(4);
 		assertThat(arguments.get("x-queue-mode")).isEqualTo("lazy");
-		assertThat(arguments.get("x-queue-master-locator")).isEqualTo("min-masters");
+		assertThat(arguments.get(Queue.X_QUEUE_LEADER_LOCATOR)).isEqualTo(LeaderLocator.minLeaders.getValue());
 		assertThat(arguments.get("x-single-active-consumer")).isEqualTo(Boolean.TRUE);
 	}
 
@@ -130,7 +130,7 @@ void testQueueArgs2() throws MalformedURLException, URISyntaxException, Interrup
 		assertThat(arguments.get("x-dead-letter-routing-key")).isEqualTo("reply.dlrk");
 		assertThat(arguments.get("x-max-priority")).isEqualTo(4);
 		assertThat(arguments.get("x-queue-mode")).isEqualTo("lazy");
-		assertThat(arguments.get("x-queue-master-locator")).isEqualTo("client-local");
+		assertThat(arguments.get(Queue.X_QUEUE_LEADER_LOCATOR)).isEqualTo(LeaderLocator.clientLocal.getValue());
 	}
 
 	@Test
@@ -148,7 +148,7 @@ void testQueueArgs3() throws MalformedURLException, URISyntaxException, Interrup
 		assertThat(arguments.get("x-dead-letter-routing-key")).isEqualTo("reply.dlrk");
 		assertThat(arguments.get("x-max-priority")).isEqualTo(4);
 		assertThat(arguments.get("x-queue-mode")).isEqualTo("lazy");
-		assertThat(arguments.get("x-queue-master-locator")).isEqualTo("random");
+		assertThat(arguments.get(Queue.X_QUEUE_LEADER_LOCATOR)).isEqualTo(LeaderLocator.random.getValue());
 
 		ExchangeInfo exchange = client.getExchange("/", "dlx.test.requestEx");
 		assertThat(exchange.getArguments().get("alternate-exchange")).isEqualTo("alternate");
@@ -294,7 +294,7 @@ public Queue allArgs1() {
 					.deadLetterRoutingKey("reply.dlrk")
 					.maxPriority(4)
 					.lazy()
-					.masterLocator(MasterLocator.minMasters)
+					.leaderLocator(LeaderLocator.minLeaders)
 					.singleActiveConsumer()
 					.build();
 		}
@@ -311,7 +311,7 @@ public Queue allArgs2() {
 					.deadLetterRoutingKey("reply.dlrk")
 					.maxPriority(4)
 					.lazy()
-					.masterLocator(MasterLocator.clientLocal)
+					.leaderLocator(LeaderLocator.clientLocal)
 					.build();
 		}
 
@@ -327,7 +327,7 @@ public Queue allArgs3() {
 					.deadLetterRoutingKey("reply.dlrk")
 					.maxPriority(4)
 					.lazy()
-					.masterLocator(MasterLocator.random)
+					.leaderLocator(LeaderLocator.random)
 					.build();
 		}
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -367,7 +367,7 @@ public void testRetry() throws Exception {
 	}
 
 	@Test
-	public void testMasterLocator() throws Exception {
+	public void testLeaderLocator() throws Exception {
 		CachingConnectionFactory cf = new CachingConnectionFactory(
 				RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
 		RabbitAdmin admin = new RabbitAdmin(cf);
@@ -376,14 +376,14 @@ public void testMasterLocator() throws Exception {
 		Client client = new Client("http://guest:guest@localhost:15672/api");
 		AnonymousQueue queue1 = queue;
 		QueueInfo info = await().until(() -> client.getQueue("/", queue1.getName()), inf -> inf != null);
-		assertThat(info.getArguments().get(Queue.X_QUEUE_MASTER_LOCATOR)).isEqualTo("client-local");
+		assertThat(info.getArguments().get(Queue.X_QUEUE_LEADER_LOCATOR)).isEqualTo("client-local");
 
 		queue = new AnonymousQueue();
-		queue.setMasterLocator(null);
+		queue.setLeaderLocator(null);
 		admin.declareQueue(queue);
 		AnonymousQueue queue2 = queue;
 		info = await().until(() -> client.getQueue("/", queue2.getName()), inf -> inf != null);
-		assertThat(info.getArguments().get(Queue.X_QUEUE_MASTER_LOCATOR)).isNull();
+		assertThat(info.getArguments().get(Queue.X_QUEUE_LEADER_LOCATOR)).isNull();
 		cf.destroy();
 	}
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/TransactionalEventListenerTests.java

@@ -112,6 +112,7 @@ public void publish() {
 			publisher.publishEvent("test");
 		}
 
+		@SuppressWarnings("serial")
 		@Bean
 		PlatformTransactionManager transactionManager(AtomicBoolean committed) {
 			return new AbstractPlatformTransactionManager() {

src/reference/asciidoc/amqp.adoc

@@ -698,12 +698,12 @@ See <<cf-pub-conf-ret>>.
 ===== Queue Affinity and the `LocalizedQueueConnectionFactory`
 
 When using HA queues in a cluster, for the best performance, you may want to connect to the physical broker
-where the master queue resides.
+where the lead queue resides.
 The `CachingConnectionFactory` can be configured with multiple broker addresses.
 This is to fail over and the client attempts to connect in order.
-The `LocalizedQueueConnectionFactory` uses the REST API provided by the admin plugin to determine on which node the queue is mastered.
+The `LocalizedQueueConnectionFactory` uses the REST API provided by the management plugin to determine which node is the lead for the queue.
 It then creates (or retrieves from a cache) a `CachingConnectionFactory` that connects to just that node.
-If the connection fails, the new master node is determined and the consumer connects to it.
+If the connection fails, the new lead node is determined and the consumer connects to it.
 The `LocalizedQueueConnectionFactory` is configured with a default connection factory, in case the physical location of the queue cannot be determined, in which case it connects as normal to the cluster.
 
 The `LocalizedQueueConnectionFactory` is a `RoutingConnectionFactory` and the `SimpleMessageListenerContainer` uses the queue names as the lookup key as discussed in <<routing-connection-factory>> above.
@@ -748,8 +748,7 @@ public LocalizedQueueConnectionFactory queueAffinityCF(
 ====
 
 Notice that the first three parameters are arrays of `addresses`, `adminUris`, and `nodes`.
-These are positional in that, when a container attempts to connect to a queue, it uses the admin API to determine on which node the queue is
-mastered and connects to the address in the same array position as that node.
+These are positional in that, when a container attempts to connect to a queue, it uses the admin API to determine which node is the lead for the queue and connects to the address in the same array position as that node.
 
 [[cf-pub-conf-ret]]
 ===== Publisher Confirms and Returns
@@ -4415,7 +4414,7 @@ A `<rabbit:queue/>` with an empty (or missing) `name` attribute always creates a
 
 See <<anonymous-queue>> to understand why `AnonymousQueue` is preferred over broker-generated queue names as well as
 how to control the format of the name.
-Starting with version 2.1, anonymous queues are declared with argument `x-queue-master-locator` set to `client-local` by default.
+Starting with version 2.1, anonymous queues are declared with argument `Queue.X_QUEUE_LEADER_LOCATOR` set to `client-local` by default.
 This ensures that the queue is declared on the node to which the application is connected.
 Declarative queues must have fixed names because they might be referenced elsewhere in the context -- such as in the
 listener shown in the following example:
@@ -4549,7 +4548,7 @@ With Spring Framework 3.2 and later, this can be declared a little more succinct
 ----
 ====
 
-When you use Java configuration, the `x-queue-master-locator` is supported as a first class property through the `setMasterLocator()` method on the `Queue` class.
+When you use Java configuration, the `Queue.X_QUEUE_LEADER_LOCATOR` argument is supported as a first class property through the `setLeaderLocator()` method on the `Queue` class.
 Starting with version 2.1, anonymous queues are declared with this property set to `client-local` by default.
 This ensures that the queue is declared on the node the application is connected to.
 
@@ -4757,7 +4756,7 @@ public Queue allArgs1() {
             .deadLetterRoutingKey("dlrk")
             .maxPriority(4)
             .lazy()
-            .masterLocator(MasterLocator.minMasters)
+            .leaderLocator(LeaderLocator.minLeaders)
             .singleActiveConsumer()
             .build();
 }
@@ -5053,9 +5052,9 @@ The third example creates names such as `custom.gen-MRBv9sqISkuCiPfOYfpo4g`.
 
 You can also provide your own naming strategy bean.
 
-Starting with version 2.1, anonymous queues are declared with argument `x-queue-master-locator` set to `client-local` by default.
+Starting with version 2.1, anonymous queues are declared with argument `Queue.X_QUEUE_LEADER_LOCATOR` set to `client-local` by default.
 This ensures that the queue is declared on the node to which the application is connected.
-You can revert to the previous behavior by calling `queue.setMasterLocator(null)` after constructing the instance.
+You can revert to the previous behavior by calling `queue.setLeaderLocator(null)` after constructing the instance.
 
 [[broker-events]]
 ==== Broker Event Listener