I have tested WSO2 MB 2.2.0 in a clustered deployment with a WS subscription to a topic. I have replicated the setup described in the project documentation.
When the first node in the cluster starts-up it gets the durable subscription queue as expected. However, the second and later nodes also try to get that durable connections and the broker denies the access throwing an exception. This is the expected behaviour from the broker part, but the topic management part of the rest of the nodes crash and thus the rest of the members of the cluster can't start up to an usable status.
Is this a known issue, an expected behaviour of the clusters? Does it means that the CarbonEventBroker component is not suitable for a clustered environment?
Here you are the log of the exception raised while launching the IService component (I guess):
TID: [0] [MB] [2015-11-16 13:11:36,670] ERROR {org.wso2.carbon.event.core.internal.builder.EventBrokerHandler} - Can not create the event broker {org.wso2.carbon.event.core.internal.builder.EventBrokerHandler}
org.wso2.carbon.event.core.exception.EventBrokerConfigurationException: Can not get the subscriptions
at org.wso2.carbon.event.core.internal.CarbonEventBroker.loadExistingSubscriptions(CarbonEventBroker.java:83)
at org.wso2.carbon.event.core.internal.CarbonEventBroker.init(CarbonEventBroker.java:67)
at org.wso2.carbon.event.core.internal.CarbonEventBrokerFactory.getEventBroker(CarbonEventBrokerFactory.java:90)
at org.wso2.carbon.event.core.internal.builder.EventBrokerBuilder.createEventBroker(EventBrokerBuilder.java:56)
at org.wso2.carbon.event.core.internal.builder.EventBrokerHandler.startEventBroker(EventBrokerHandler.java:57)
at org.wso2.carbon.event.core.internal.builder.EventBrokerHandler.notifyStart(EventBrokerHandler.java:96)
at org.wso2.carbon.andes.internal.QpidServiceComponent.activate(QpidServiceComponent.java:258)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at org.eclipse.equinox.internal.ds.model.ServiceComponent.activate(ServiceComponent.java:260)
at org.eclipse.equinox.internal.ds.model.ServiceComponentProp.activate(ServiceComponentProp.java:146)
at org.eclipse.equinox.internal.ds.model.ServiceComponentProp.build(ServiceComponentProp.java:347)
at org.eclipse.equinox.internal.ds.InstanceProcess.buildComponent(InstanceProcess.java:620)
at org.eclipse.equinox.internal.ds.InstanceProcess.buildComponents(InstanceProcess.java:197)
at org.eclipse.equinox.internal.ds.Resolver.getEligible(Resolver.java:343)
at org.eclipse.equinox.internal.ds.SCRManager.serviceChanged(SCRManager.java:222)
at org.eclipse.osgi.internal.serviceregistry.FilteredServiceListener.serviceChanged(FilteredServiceListener.java:107)
at org.eclipse.osgi.framework.internal.core.BundleContextImpl.dispatchEvent(BundleContextImpl.java:861)
at org.eclipse.osgi.framework.eventmgr.EventManager.dispatchEvent(EventManager.java:230)
at org.eclipse.osgi.framework.eventmgr.ListenerQueue.dispatchEventSynchronous(ListenerQueue.java:148)
at org.eclipse.osgi.internal.serviceregistry.ServiceRegistry.publishServiceEventPrivileged(ServiceRegistry.java:819)
at org.eclipse.osgi.internal.serviceregistry.ServiceRegistry.publishServiceEvent(ServiceRegistry.java:771)
at org.eclipse.osgi.internal.serviceregistry.ServiceRegistrationImpl.register(ServiceRegistrationImpl.java:130)
at org.eclipse.osgi.internal.serviceregistry.ServiceRegistry.registerService(ServiceRegistry.java:214)
at org.eclipse.osgi.framework.internal.core.BundleContextImpl.registerService(BundleContextImpl.java:433)
at org.eclipse.osgi.framework.internal.core.BundleContextImpl.registerService(BundleContextImpl.java:451)
at org.wso2.carbon.event.core.internal.builder.EventBrokerBuilderDS.activate(EventBrokerBuilderDS.java:69)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at org.eclipse.equinox.internal.ds.model.ServiceComponent.activate(ServiceComponent.java:260)
at org.eclipse.equinox.internal.ds.model.ServiceComponentProp.activate(ServiceComponentProp.java:146)
at org.eclipse.equinox.internal.ds.model.ServiceComponentProp.build(ServiceComponentProp.java:347)
at org.eclipse.equinox.internal.ds.InstanceProcess.buildComponent(InstanceProcess.java:620)
at org.eclipse.equinox.internal.ds.InstanceProcess.buildComponents(InstanceProcess.java:197)
at org.eclipse.equinox.internal.ds.Resolver.getEligible(Resolver.java:343)
at org.eclipse.equinox.internal.ds.SCRManager.serviceChanged(SCRManager.java:222)
at org.eclipse.osgi.internal.serviceregistry.FilteredServiceListener.serviceChanged(FilteredServiceListener.java:107)
at org.eclipse.osgi.framework.internal.core.BundleContextImpl.dispatchEvent(BundleContextImpl.java:861)
at org.eclipse.osgi.framework.eventmgr.EventManager.dispatchEvent(EventManager.java:230)
at org.eclipse.osgi.framework.eventmgr.ListenerQueue.dispatchEventSynchronous(ListenerQueue.java:148)
at org.eclipse.osgi.internal.serviceregistry.ServiceRegistry.publishServiceEventPrivileged(ServiceRegistry.java:819)
at org.eclipse.osgi.internal.serviceregistry.ServiceRegistry.publishServiceEvent(ServiceRegistry.java:771)
at org.eclipse.osgi.internal.serviceregistry.ServiceRegistrationImpl.register(ServiceRegistrationImpl.java:130)
at org.eclipse.osgi.internal.serviceregistry.ServiceRegistry.registerService(ServiceRegistry.java:214)
at org.eclipse.osgi.framework.internal.core.BundleContextImpl.registerService(BundleContextImpl.java:433)
at org.eclipse.osgi.framework.internal.core.BundleContextImpl.registerService(BundleContextImpl.java:451)
at org.wso2.carbon.core.init.CarbonServerManager.initializeCarbon(CarbonServerManager.java:517)
at org.wso2.carbon.core.init.CarbonServerManager.start(CarbonServerManager.java:219)
at org.wso2.carbon.core.internal.CarbonCoreServiceComponent.activate(CarbonCoreServiceComponent.java:77)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at org.eclipse.equinox.internal.ds.model.ServiceComponent.activate(ServiceComponent.java:260)
at org.eclipse.equinox.internal.ds.model.ServiceComponentProp.activate(ServiceComponentProp.java:146)
at org.eclipse.equinox.internal.ds.model.ServiceComponentProp.build(ServiceComponentProp.java:347)
at org.eclipse.equinox.internal.ds.InstanceProcess.buildComponent(InstanceProcess.java:620)
at org.eclipse.equinox.internal.ds.InstanceProcess.buildComponents(InstanceProcess.java:197)
at org.eclipse.equinox.internal.ds.Resolver.getEligible(Resolver.java:343)
at org.eclipse.equinox.internal.ds.SCRManager.serviceChanged(SCRManager.java:222)
at org.eclipse.osgi.internal.serviceregistry.FilteredServiceListener.serviceChanged(FilteredServiceListener.java:107)
at org.eclipse.osgi.framework.internal.core.BundleContextImpl.dispatchEvent(BundleContextImpl.java:861)
at org.eclipse.osgi.framework.eventmgr.EventManager.dispatchEvent(EventManager.java:230)
at org.eclipse.osgi.framework.eventmgr.ListenerQueue.dispatchEventSynchronous(ListenerQueue.java:148)
at org.eclipse.osgi.internal.serviceregistry.ServiceRegistry.publishServiceEventPrivileged(ServiceRegistry.java:819)
at org.eclipse.osgi.internal.serviceregistry.ServiceRegistry.publishServiceEvent(ServiceRegistry.java:771)
at org.eclipse.osgi.internal.serviceregistry.ServiceRegistrationImpl.register(ServiceRegistrationImpl.java:130)
at org.eclipse.osgi.internal.serviceregistry.ServiceRegistry.registerService(ServiceRegistry.java:214)
at org.eclipse.osgi.framework.internal.core.BundleContextImpl.registerService(BundleContextImpl.java:433)
at org.eclipse.equinox.http.servlet.internal.Activator.registerHttpService(Activator.java:81)
at org.eclipse.equinox.http.servlet.internal.Activator.addProxyServlet(Activator.java:60)
at org.eclipse.equinox.http.servlet.internal.ProxyServlet.init(ProxyServlet.java:40)
at org.wso2.carbon.tomcat.ext.servlet.DelegationServlet.init(DelegationServlet.java:38)
at org.apache.catalina.core.StandardWrapper.initServlet(StandardWrapper.java:1267)
at org.apache.catalina.core.StandardWrapper.loadServlet(StandardWrapper.java:1186)
at org.apache.catalina.core.StandardWrapper.load(StandardWrapper.java:1081)
at org.apache.catalina.core.StandardContext.loadOnStartup(StandardContext.java:5027)
at org.apache.catalina.core.StandardContext.startInternal(StandardContext.java:5314)
at org.apache.catalina.util.LifecycleBase.start(LifecycleBase.java:150)
at org.apache.catalina.core.ContainerBase$StartChild.call(ContainerBase.java:1559)
at org.apache.catalina.core.ContainerBase$StartChild.call(ContainerBase.java:1549)
at java.util.concurrent.FutureTask.run(FutureTask.java:262)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.wso2.carbon.event.core.exception.EventBrokerException: Can not subscribe to topic /msg/EVT-01 Error registering consumer: org.wso2.andes.AMQSecurityException: Cannot subscribe to queue carbon:bc4688e3fb936491cf1f524905ffbd9ef816e80bea175512 as it already has an existing exclusive consumer [error code 403: access refused]
at org.wso2.carbon.event.core.internal.delivery.jms.JMSDeliveryManager.subscribe(JMSDeliveryManager.java:136)
at org.wso2.carbon.event.core.internal.CarbonEventBroker.loadExistingSubscriptions(CarbonEventBroker.java:79)
... 88 more
Caused by: javax.jms.JMSException: Error registering consumer: org.wso2.andes.AMQSecurityException: Cannot subscribe to queue carbon:bc4688e3fb936491cf1f524905ffbd9ef816e80bea175512 as it already has an existing exclusive consumer [error code 403: access refused]
at org.wso2.andes.client.AMQSession$5.execute(AMQSession.java:2084)
at org.wso2.andes.client.AMQSession$5.execute(AMQSession.java:2027)
at org.wso2.andes.client.AMQConnectionDelegate_8_0.executeRetrySupport(AMQConnectionDelegate_8_0.java:315)
at org.wso2.andes.client.AMQConnection.executeRetrySupport(AMQConnection.java:621)
at org.wso2.andes.client.failover.FailoverRetrySupport.execute(FailoverRetrySupport.java:102)
at org.wso2.andes.client.AMQSession.createConsumerImpl(AMQSession.java:2025)
at org.wso2.andes.client.AMQSession.createConsumer(AMQSession.java:1028)
at org.wso2.andes.client.AMQSession.createDurableSubscriber(AMQSession.java:1177)
at org.wso2.andes.client.AMQSession.createDurableSubscriber(AMQSession.java:1077)
at org.wso2.andes.client.AMQTopicSessionAdaptor.createDurableSubscriber(AMQTopicSessionAdaptor.java:73)
at org.wso2.carbon.event.core.internal.delivery.jms.JMSDeliveryManager.subscribe(JMSDeliveryManager.java:130)
... 89 more
Caused by: org.wso2.andes.AMQSecurityException: Cannot subscribe to queue carbon:bc4688e3fb936491cf1f524905ffbd9ef816e80bea175512 as it already has an existing exclusive consumer [error code 403: access refused]
at org.wso2.andes.client.handler.ConnectionCloseMethodHandler.methodReceived(ConnectionCloseMethodHandler.java:85)
at org.wso2.andes.client.handler.ClientMethodDispatcherImpl.dispatchConnectionClose(ClientMethodDispatcherImpl.java:192)
at org.wso2.andes.framing.amqp_0_91.ConnectionCloseBodyImpl.execute(ConnectionCloseBodyImpl.java:140)
at org.wso2.andes.client.state.AMQStateManager.methodReceived(AMQStateManager.java:111)
at org.wso2.andes.client.protocol.AMQProtocolHandler.methodBodyReceived(AMQProtocolHandler.java:515)
at org.wso2.andes.client.protocol.AMQProtocolSession.methodFrameReceived(AMQProtocolSession.java:461)
at org.wso2.andes.framing.AMQMethodBodyImpl.handle(AMQMethodBodyImpl.java:96)
at org.wso2.andes.client.protocol.AMQProtocolHandler$2.run(AMQProtocolHandler.java:466)
at org.wso2.andes.pool.Job.processAll(Job.java:109)
at org.wso2.andes.pool.Job.run(Job.java:157)
... 3 more
TID: [0] [MB] [2015-11-16 13:11:36,671] ERROR {org.wso2.andes.client.AMQConnection} - Throwable Received but no listener set: org.wso2.andes.AMQException: Cannot subscribe to queue carbon:bc4688e3fb936491cf1f524905ffbd9ef816e80bea175512 as it already has an existing exclusive consumer [error code 403: access refused] {org.wso2.andes.client.AMQConnection}