NO-JIRA Using force option on deleteTopics and deleteQueues on JoramTests

This commit is contained in:
Clebert Suconic 2018-02-27 15:50:52 -05:00
parent c6028c779f
commit 158eb9d3bf
9 changed files with 93 additions and 47 deletions

View File

@ -438,7 +438,6 @@ public interface ActiveMQServerControl {
@Attribute(desc = "global maximum limit for in-memory messages, in bytes") @Attribute(desc = "global maximum limit for in-memory messages, in bytes")
long getGlobalMaxSize(); long getGlobalMaxSize();
/** /**
* Returns the memory used by all the addresses on broker for in-memory messages * Returns the memory used by all the addresses on broker for in-memory messages
*/ */
@ -457,15 +456,19 @@ public interface ActiveMQServerControl {
@Operation(desc = "create an address", impact = MBeanOperationInfo.ACTION) @Operation(desc = "create an address", impact = MBeanOperationInfo.ACTION)
String createAddress(@Parameter(name = "name", desc = "The name of the address") String name, String createAddress(@Parameter(name = "name", desc = "The name of the address") String name,
@Parameter(name = "routingTypes", desc = "Comma separated list of Routing Types (anycast/multicast)") String routingTypes) throws Exception; @Parameter(name = "routingTypes", desc = "Comma separated list of Routing Types (anycast/multicast)") String routingTypes) throws Exception;
@Operation(desc = "update an address", impact = MBeanOperationInfo.ACTION) @Operation(desc = "update an address", impact = MBeanOperationInfo.ACTION)
String updateAddress(@Parameter(name = "name", desc = "The name of the address") String name, String updateAddress(@Parameter(name = "name", desc = "The name of the address") String name,
@Parameter(name = "routingTypes", desc = "Comma separated list of Routing Types (anycast/multicast)") String routingTypes) throws Exception; @Parameter(name = "routingTypes", desc = "Comma separated list of Routing Types (anycast/multicast)") String routingTypes) throws Exception;
@Operation(desc = "delete an address", impact = MBeanOperationInfo.ACTION) @Operation(desc = "delete an address", impact = MBeanOperationInfo.ACTION)
void deleteAddress(@Parameter(name = "name", desc = "The name of the address") String name) throws Exception; void deleteAddress(@Parameter(name = "name", desc = "The name of the address") String name) throws Exception;
@Operation(desc = "delete an address", impact = MBeanOperationInfo.ACTION)
void deleteAddress(@Parameter(name = "name", desc = "The name of the address") String name,
@Parameter(name = "force", desc = "Force consumers and queues out") boolean force) throws Exception;
/** /**
* Create a durable queue. * Create a durable queue.
* <br> * <br>
@ -481,7 +484,6 @@ public interface ActiveMQServerControl {
void createQueue(@Parameter(name = "address", desc = "Address of the queue") String address, void createQueue(@Parameter(name = "address", desc = "Address of the queue") String address,
@Parameter(name = "name", desc = "Name of the queue") String name) throws Exception; @Parameter(name = "name", desc = "Name of the queue") String name) throws Exception;
/** /**
* Create a durable queue. * Create a durable queue.
* <br> * <br>
@ -489,8 +491,8 @@ public interface ActiveMQServerControl {
* <br> * <br>
* This method throws a {@link org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException}) exception if the queue already exits. * This method throws a {@link org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException}) exception if the queue already exits.
* *
* @param address address to bind the queue to * @param address address to bind the queue to
* @param name name of the queue * @param name name of the queue
* @param routingType The routing type used for this address, MULTICAST or ANYCAST * @param routingType The routing type used for this address, MULTICAST or ANYCAST
*/ */
@Operation(desc = "Create a queue with the specified address", impact = MBeanOperationInfo.ACTION) @Operation(desc = "Create a queue with the specified address", impact = MBeanOperationInfo.ACTION)
@ -498,7 +500,6 @@ public interface ActiveMQServerControl {
@Parameter(name = "name", desc = "Name of the queue") String name, @Parameter(name = "name", desc = "Name of the queue") String name,
@Parameter(name = "routingType", desc = "The routing type used for this address, MULTICAST or ANYCAST") String routingType) throws Exception; @Parameter(name = "routingType", desc = "The routing type used for this address, MULTICAST or ANYCAST") String routingType) throws Exception;
/** /**
* Create a queue. * Create a queue.
* <br> * <br>
@ -523,9 +524,9 @@ public interface ActiveMQServerControl {
* <br> * <br>
* This method throws a {@link org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException}) exception if the queue already exits. * This method throws a {@link org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException}) exception if the queue already exits.
* *
* @param address address to bind the queue to * @param address address to bind the queue to
* @param name name of the queue * @param name name of the queue
* @param durable whether the queue is durable * @param durable whether the queue is durable
* @param routingType The routing type used for this address, MULTICAST or ANYCAST * @param routingType The routing type used for this address, MULTICAST or ANYCAST
*/ */
@Operation(desc = "Create a queue with the specified address, name and durability", impact = MBeanOperationInfo.ACTION) @Operation(desc = "Create a queue with the specified address, name and durability", impact = MBeanOperationInfo.ACTION)
@ -559,10 +560,10 @@ public interface ActiveMQServerControl {
* <br> * <br>
* This method throws a {@link org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException}) exception if the queue already exits. * This method throws a {@link org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException}) exception if the queue already exits.
* *
* @param address address to bind the queue to * @param address address to bind the queue to
* @param name name of the queue * @param name name of the queue
* @param filter of the queue * @param filter of the queue
* @param durable whether the queue is durable * @param durable whether the queue is durable
* @param routingType The routing type used for this address, MULTICAST or ANYCAST * @param routingType The routing type used for this address, MULTICAST or ANYCAST
*/ */
@Operation(desc = "Create a queue", impact = MBeanOperationInfo.ACTION) @Operation(desc = "Create a queue", impact = MBeanOperationInfo.ACTION)
@ -572,7 +573,6 @@ public interface ActiveMQServerControl {
@Parameter(name = "durable", desc = "Is the queue durable?") boolean durable, @Parameter(name = "durable", desc = "Is the queue durable?") boolean durable,
@Parameter(name = "routingType", desc = "The routing type used for this address, MULTICAST or ANYCAST") String routingType) throws Exception; @Parameter(name = "routingType", desc = "The routing type used for this address, MULTICAST or ANYCAST") String routingType) throws Exception;
/** /**
* Create a queue. * Create a queue.
* <br> * <br>
@ -580,33 +580,33 @@ public interface ActiveMQServerControl {
* <br> * <br>
* This method throws a {@link org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException}) exception if the queue already exits. * This method throws a {@link org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException}) exception if the queue already exits.
* *
* @param address address to bind the queue to * @param address address to bind the queue to
* @param routingType the routing type used for this address, {@code MULTICAST} or {@code ANYCAST} * @param routingType the routing type used for this address, {@code MULTICAST} or {@code ANYCAST}
* @param name name of the queue * @param name name of the queue
* @param filterStr filter of the queue * @param filterStr filter of the queue
* @param durable is the queue durable? * @param durable is the queue durable?
* @param maxConsumers the maximum number of consumers allowed on this queue at any one time * @param maxConsumers the maximum number of consumers allowed on this queue at any one time
* @param purgeOnNoConsumers delete this queue when the last consumer disconnects * @param purgeOnNoConsumers delete this queue when the last consumer disconnects
* @param autoCreateAddress create an address with default values should a matching address not be found * @param autoCreateAddress create an address with default values should a matching address not be found
* @return a textual summary of the queue * @return a textual summary of the queue
* @throws Exception * @throws Exception
*/ */
@Operation(desc = "Create a queue", impact = MBeanOperationInfo.ACTION) @Operation(desc = "Create a queue", impact = MBeanOperationInfo.ACTION)
String createQueue(@Parameter(name = "address", desc = "Address of the queue") String address, String createQueue(@Parameter(name = "address", desc = "Address of the queue") String address,
@Parameter(name = "routingType", desc = "The routing type used for this address, MULTICAST or ANYCAST") String routingType, @Parameter(name = "routingType", desc = "The routing type used for this address, MULTICAST or ANYCAST") String routingType,
@Parameter(name = "name", desc = "Name of the queue") String name, @Parameter(name = "name", desc = "Name of the queue") String name,
@Parameter(name = "filter", desc = "Filter of the queue") String filterStr, @Parameter(name = "filter", desc = "Filter of the queue") String filterStr,
@Parameter(name = "durable", desc = "Is the queue durable?") boolean durable, @Parameter(name = "durable", desc = "Is the queue durable?") boolean durable,
@Parameter(name = "maxConsumers", desc = "The maximum number of consumers allowed on this queue at any one time") int maxConsumers, @Parameter(name = "maxConsumers", desc = "The maximum number of consumers allowed on this queue at any one time") int maxConsumers,
@Parameter(name = "purgeOnNoConsumers", desc = "Delete this queue when the last consumer disconnects") boolean purgeOnNoConsumers, @Parameter(name = "purgeOnNoConsumers", desc = "Delete this queue when the last consumer disconnects") boolean purgeOnNoConsumers,
@Parameter(name = "autoCreateAddress", desc = "Create an address with default values should a matching address not be found") boolean autoCreateAddress) throws Exception; @Parameter(name = "autoCreateAddress", desc = "Create an address with default values should a matching address not be found") boolean autoCreateAddress) throws Exception;
/** /**
* Update a queue. * Update a queue.
* *
* @param name name of the queue * @param name name of the queue
* @param routingType the routing type used for this address, {@code MULTICAST} or {@code ANYCAST} * @param routingType the routing type used for this address, {@code MULTICAST} or {@code ANYCAST}
* @param maxConsumers the maximum number of consumers allowed on this queue at any one time * @param maxConsumers the maximum number of consumers allowed on this queue at any one time
* @param purgeOnNoConsumers delete this queue when the last consumer disconnects * @param purgeOnNoConsumers delete this queue when the last consumer disconnects
* @return a textual summary of the queue * @return a textual summary of the queue
* @throws Exception * @throws Exception
@ -616,13 +616,12 @@ public interface ActiveMQServerControl {
@Parameter(name = "maxConsumers", desc = "The maximum number of consumers allowed on this queue at any one time") Integer maxConsumers, @Parameter(name = "maxConsumers", desc = "The maximum number of consumers allowed on this queue at any one time") Integer maxConsumers,
@Parameter(name = "purgeOnNoConsumers", desc = "Delete this queue when the last consumer disconnects") Boolean purgeOnNoConsumers) throws Exception; @Parameter(name = "purgeOnNoConsumers", desc = "Delete this queue when the last consumer disconnects") Boolean purgeOnNoConsumers) throws Exception;
/** /**
* Update a queue. * Update a queue.
* *
* @param name name of the queue * @param name name of the queue
* @param routingType the routing type used for this address, {@code MULTICAST} or {@code ANYCAST} * @param routingType the routing type used for this address, {@code MULTICAST} or {@code ANYCAST}
* @param maxConsumers the maximum number of consumers allowed on this queue at any one time * @param maxConsumers the maximum number of consumers allowed on this queue at any one time
* @param purgeOnNoConsumers delete this queue when the last consumer disconnects * @param purgeOnNoConsumers delete this queue when the last consumer disconnects
* @return a textual summary of the queue * @return a textual summary of the queue
* @throws Exception * @throws Exception
@ -633,7 +632,6 @@ public interface ActiveMQServerControl {
@Parameter(name = "purgeOnNoConsumers", desc = "Delete this queue when the last consumer disconnects") Boolean purgeOnNoConsumers, @Parameter(name = "purgeOnNoConsumers", desc = "Delete this queue when the last consumer disconnects") Boolean purgeOnNoConsumers,
@Parameter(name = "exclusive", desc = "If the queue should route exclusively to one consumer") Boolean exclusive) throws Exception; @Parameter(name = "exclusive", desc = "If the queue should route exclusively to one consumer") Boolean exclusive) throws Exception;
/** /**
* Deploy a durable queue. * Deploy a durable queue.
* <br> * <br>
@ -689,7 +687,6 @@ public interface ActiveMQServerControl {
@Parameter(name = "removeConsumers", desc = "Remove consumers of this queue") boolean removeConsumers, @Parameter(name = "removeConsumers", desc = "Remove consumers of this queue") boolean removeConsumers,
@Parameter(name = "autoDeleteAddress", desc = "Automatically delete the address if this was the last queue") boolean autoDeleteAddress) throws Exception; @Parameter(name = "autoDeleteAddress", desc = "Automatically delete the address if this was the last queue") boolean autoDeleteAddress) throws Exception;
/** /**
* Enables message counters for this server. * Enables message counters for this server.
*/ */
@ -987,6 +984,7 @@ public interface ActiveMQServerControl {
@Parameter(desc = "allow auto-created queues to be deleted automatically", name = "autoDeleteJmsQueues") boolean autoDeleteJmsQueues, @Parameter(desc = "allow auto-created queues to be deleted automatically", name = "autoDeleteJmsQueues") boolean autoDeleteJmsQueues,
@Parameter(desc = "allow topics to be created automatically", name = "autoCreateJmsTopics") boolean autoCreateJmsTopics, @Parameter(desc = "allow topics to be created automatically", name = "autoCreateJmsTopics") boolean autoCreateJmsTopics,
@Parameter(desc = "allow auto-created topics to be deleted automatically", name = "autoDeleteJmsTopics") boolean autoDeleteJmsTopics) throws Exception; @Parameter(desc = "allow auto-created topics to be deleted automatically", name = "autoDeleteJmsTopics") boolean autoDeleteJmsTopics) throws Exception;
/** /**
* adds a new address setting for a specific address * adds a new address setting for a specific address
*/ */
@ -1213,8 +1211,8 @@ public interface ActiveMQServerControl {
@Operation(desc = "Search for Consumers", impact = MBeanOperationInfo.INFO) @Operation(desc = "Search for Consumers", impact = MBeanOperationInfo.INFO)
String listConsumers(@Parameter(name = "Options") String options, String listConsumers(@Parameter(name = "Options") String options,
@Parameter(name = "Page Number") int page, @Parameter(name = "Page Number") int page,
@Parameter(name = "Page Size") int pageSize) throws Exception; @Parameter(name = "Page Size") int pageSize) throws Exception;
@Operation(desc = "Search for Consumers", impact = MBeanOperationInfo.INFO) @Operation(desc = "Search for Consumers", impact = MBeanOperationInfo.INFO)
String listProducers(@Parameter(name = "Options") String options, String listProducers(@Parameter(name = "Options") String options,

View File

@ -722,13 +722,19 @@ public class ActiveMQServerControlImpl extends AbstractControl implements Active
} }
} }
@Override @Override
public void deleteAddress(String name) throws Exception { public void deleteAddress(String name) throws Exception {
deleteAddress(name, false);
}
@Override
public void deleteAddress(String name, boolean force) throws Exception {
checkStarted(); checkStarted();
clearIO(); clearIO();
try { try {
server.removeAddressInfo(new SimpleString(name), null); server.removeAddressInfo(new SimpleString(name), null, force);
} catch (ActiveMQException e) { } catch (ActiveMQException e) {
throw new IllegalStateException(e.getMessage()); throw new IllegalStateException(e.getMessage());
} finally { } finally {

View File

@ -58,6 +58,8 @@ public interface PostOffice extends ActiveMQComponent {
AddressInfo removeAddressInfo(SimpleString address) throws Exception; AddressInfo removeAddressInfo(SimpleString address) throws Exception;
AddressInfo removeAddressInfo(SimpleString address, boolean force) throws Exception;
AddressInfo getAddressInfo(SimpleString address); AddressInfo getAddressInfo(SimpleString address);
AddressInfo updateAddressInfo(SimpleString addressName, EnumSet<RoutingType> routingTypes) throws Exception; AddressInfo updateAddressInfo(SimpleString addressName, EnumSet<RoutingType> routingTypes) throws Exception;

View File

@ -544,16 +544,31 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
} }
} }
@Override @Override
public AddressInfo removeAddressInfo(SimpleString address) throws Exception { public AddressInfo removeAddressInfo(SimpleString address) throws Exception {
return removeAddressInfo(address, false);
}
@Override
public AddressInfo removeAddressInfo(SimpleString address, boolean force) throws Exception {
synchronized (addressLock) { synchronized (addressLock) {
if (server.hasBrokerPlugins()) { if (server.hasBrokerPlugins()) {
server.callBrokerPlugins(plugin -> plugin.beforeRemoveAddress(address)); server.callBrokerPlugins(plugin -> plugin.beforeRemoveAddress(address));
} }
final Bindings bindingsForAddress = getDirectBindings(address); final Bindings bindingsForAddress = getDirectBindings(address);
if (bindingsForAddress.getBindings().size() > 0) { if (force) {
throw ActiveMQMessageBundle.BUNDLE.addressHasBindings(address); for (Binding binding : bindingsForAddress.getBindings()) {
if (binding instanceof QueueBinding) {
((QueueBinding)binding).getQueue().deleteQueue(true);
}
}
} else {
if (bindingsForAddress.getBindings().size() > 0) {
throw ActiveMQMessageBundle.BUNDLE.addressHasBindings(address);
}
} }
managementService.unregisterAddress(address); managementService.unregisterAddress(address);
final AddressInfo addressInfo = addressManager.removeAddressInfo(address); final AddressInfo addressInfo = addressManager.removeAddressInfo(address);

View File

@ -511,7 +511,6 @@ public interface ActiveMQServer extends ServiceComponent {
* @throws Exception * @throws Exception
*/ */
AddressInfo addOrUpdateAddressInfo(AddressInfo addressInfo) throws Exception; AddressInfo addOrUpdateAddressInfo(AddressInfo addressInfo) throws Exception;
/** /**
* Remove an {@code AddressInfo} from the broker. * Remove an {@code AddressInfo} from the broker.
* *
@ -521,6 +520,16 @@ public interface ActiveMQServer extends ServiceComponent {
*/ */
void removeAddressInfo(SimpleString address, SecurityAuth auth) throws Exception; void removeAddressInfo(SimpleString address, SecurityAuth auth) throws Exception;
/**
* Remove an {@code AddressInfo} from the broker.
*
* @param address the {@code AddressInfo} to remove
* @param auth authorization information; {@code null} is valid
* @param force It will disconnect everything from the address including queues and consumers
* @throws Exception
*/
void removeAddressInfo(SimpleString address, SecurityAuth auth, boolean force) throws Exception;
String getInternalNamingPrefix(); String getInternalNamingPrefix();
} }

View File

@ -2661,14 +2661,20 @@ public class ActiveMQServerImpl implements ActiveMQServer {
return getAddressInfo(addressInfo.getName()); return getAddressInfo(addressInfo.getName());
} }
@Override @Override
public void removeAddressInfo(final SimpleString address, final SecurityAuth auth) throws Exception { public void removeAddressInfo(final SimpleString address, final SecurityAuth auth) throws Exception {
removeAddressInfo(address, auth, false);
}
@Override
public void removeAddressInfo(final SimpleString address, final SecurityAuth auth, boolean force) throws Exception {
if (auth != null) { if (auth != null) {
securityStore.check(address, CheckType.DELETE_ADDRESS, auth); securityStore.check(address, CheckType.DELETE_ADDRESS, auth);
} }
AddressInfo addressInfo = getAddressInfo(address); AddressInfo addressInfo = getAddressInfo(address);
if (postOffice.removeAddressInfo(address) == null) { if (postOffice.removeAddressInfo(address, force) == null) {
throw ActiveMQMessageBundle.BUNDLE.addressDoesNotExist(address); throw ActiveMQMessageBundle.BUNDLE.addressDoesNotExist(address);
} }

View File

@ -150,6 +150,11 @@ public class ActiveMQServerControlUsingCoreTest extends ActiveMQServerControlTes
proxy.invokeOperation("deleteAddress", name); proxy.invokeOperation("deleteAddress", name);
} }
@Override
public void deleteAddress(@Parameter(name = "name", desc = "The name of the address") String name, @Parameter(name = "force", desc = "Force everything out!") boolean force) throws Exception {
proxy.invokeOperation("deleteAddress", name, force);
}
@Override @Override
public void createQueue(final String address, public void createQueue(final String address,
final String name, final String name,

View File

@ -130,7 +130,7 @@ public class AbstractAdmin implements Admin {
public void deleteQueue(final String name) { public void deleteQueue(final String name) {
Boolean result; Boolean result;
try { try {
invokeSyncOperation(ResourceNames.BROKER, "destroyQueue", name); invokeSyncOperation(ResourceNames.BROKER, "destroyQueue", name, true);
} catch (Exception e) { } catch (Exception e) {
throw new IllegalStateException(e); throw new IllegalStateException(e);
} }
@ -159,7 +159,7 @@ public class AbstractAdmin implements Admin {
public void deleteTopic(final String name) { public void deleteTopic(final String name) {
Boolean result; Boolean result;
try { try {
invokeSyncOperation(ResourceNames.BROKER, "deleteAddress", name); invokeSyncOperation(ResourceNames.BROKER, "deleteAddress", name, Boolean.TRUE);
} catch (Exception e) { } catch (Exception e) {
throw new IllegalStateException(e); throw new IllegalStateException(e);
} }

View File

@ -95,6 +95,11 @@ public class FakePostOffice implements PostOffice {
} }
@Override
public AddressInfo removeAddressInfo(SimpleString address, boolean force) throws Exception {
return null;
}
@Override @Override
public boolean addAddressInfo(AddressInfo addressInfo) { public boolean addAddressInfo(AddressInfo addressInfo) {
return false; return false;