[hornetq-commits] JBoss hornetq SVN: r12282 - in branches/Branch_2_2_AS7/src/main/org/hornetq: jms/server/recovery and 1 other directories.

Author: clebert.suconic(a)jboss.com Date: 2012-03-09 16:11:33 -0500 (Fri, 09 Mar 2012) New Revision: 12282 Added: branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQRegistryBase.java Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/client/HornetQClient.java branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQResourceRecovery.java branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/RecoveryRegistry.java branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/XARecoveryConfig.java branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java Log: JBPAPP-8366 & JBPAPP-8377 - fixing leaks and duplicated resource on TM Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/client/HornetQClient.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/client/HornetQClient.java 2012-03-09 21:09:17 UTC (rev 12281) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/client/HornetQClient.java 2012-03-09 21:11:33 UTC (rev 12282) @@ -113,6 +113,18 @@ } /** + * Create a ServerLocator which creates session factories using a static list of transportConfigurations, the ServerLocator is not updated automatically + * as the cluster topology changes, and no HA backup information is propagated to the client + * + * @param transportConfigurations + * @return the ServerLocator + */ + public static ServerLocator createServerLocator(final boolean ha, TransportConfiguration... transportConfigurations) + { + return new ServerLocatorImpl(ha, transportConfigurations); + } + + /** * Create a ServerLocator which creates session factories from a set of live servers, no HA backup information is propagated to the client * * The UDP address and port are used to listen for live servers in the cluster @@ -127,6 +139,20 @@ } /** + * Create a ServerLocator which creates session factories from a set of live servers, no HA backup information is propagated to the client + * + * The UDP address and port are used to listen for live servers in the cluster + * + * @param discoveryAddress The UDP group address to listen for updates + * @param discoveryPort the UDP port to listen for updates + * @return the ServerLocator + */ + public static ServerLocator createServerLocator(final boolean ha, final DiscoveryGroupConfiguration groupConfiguration) + { + return new ServerLocatorImpl(ha, groupConfiguration); + } + + /** * Create a ServerLocator which will receive cluster topology updates from the cluster as servers leave or join and new backups are appointed or removed. * The initial list of servers supplied in this method is simply to make an initial connection to the cluster, once that connection is made, up to date * cluster topology information is downloaded and automatically updated whenever the cluster topology changes. If the topology includes backup servers Added: branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQRegistryBase.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQRegistryBase.java (rev 0) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQRegistryBase.java 2012-03-09 21:11:33 UTC (rev 12282) @@ -0,0 +1,102 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat licenses this file to you 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 + * http://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.hornetq.jms.server.recovery; + +import java.util.HashMap; + +import org.hornetq.core.logging.Logger; +import org.jboss.tm.XAResourceRecoveryRegistry; + +/** + * This class is a base class for the integration layer where + * we verify if a given connection factory already have a recovery registered + * + * @author Clebert + * + * + */ +public abstract class HornetQRegistryBase implements RecoveryRegistry +{ + // Constants ----------------------------------------------------- + + private static final Logger log = Logger.getLogger(HornetQRegistryBase.class); + + // Attributes ---------------------------------------------------- + + private static HashMap<XARecoveryConfig, HornetQResourceRecovery> configSet = new HashMap<XARecoveryConfig, HornetQResourceRecovery>(); + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + public abstract XAResourceRecoveryRegistry getTMRegistry(); + + public HornetQResourceRecovery register(final HornetQResourceRecovery resourceRecovery) + { + synchronized (configSet) + { + HornetQResourceRecovery recovery = configSet.get(resourceRecovery.getConfig()); + + if (recovery == null) + { + recovery = resourceRecovery; + if (log.isDebugEnabled()) + { + log.debug("Registering a new recovery for " + recovery.getConfig() + ", recovery = " + resourceRecovery); + } + configSet.put(resourceRecovery.getConfig(), resourceRecovery); + getTMRegistry().addXAResourceRecovery(recovery); + } + else + { + if (log.isDebugEnabled()) + { + log.debug("Return pre-existent recovery=" + recovery + " for configuration = " + resourceRecovery.getConfig()); + } + } + recovery.incrementUsage(); + return recovery; + } + } + + + + public void unRegister(final HornetQResourceRecovery resourceRecovery) + { + synchronized (configSet) + { + HornetQResourceRecovery recFound = configSet.get(resourceRecovery.getConfig()); + + if (recFound != null && recFound.decrementUsage() == 0) + { + if (log.isDebugEnabled()) + { + log.debug("Removing recovery information for " + recFound + " as all the deployments were already removed"); + } + getTMRegistry().removeXAResourceRecovery(recFound); + configSet.remove(resourceRecovery); + } + } + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + // Inner classes ------------------------------------------------- + +} Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQResourceRecovery.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQResourceRecovery.java 2012-03-09 21:09:17 UTC (rev 12281) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQResourceRecovery.java 2012-03-09 21:11:33 UTC (rev 12282) @@ -32,12 +32,15 @@ public class HornetQResourceRecovery implements XAResourceRecovery { private final XARecoveryConfig config; - private XAResource[] xaResources; + private final XAResource[] xaResources; + + private int usage; + public HornetQResourceRecovery(XARecoveryConfig config) { this.config = config; - xaResources = new XAResource[]{new HornetQXAResourceWrapper(config)}; + this.xaResources = new XAResource[] { new HornetQXAResourceWrapper(config) }; } public XAResource[] getXAResources() @@ -45,15 +48,36 @@ return xaResources; } + public XARecoveryConfig getConfig() + { + return config; + } + + /** we may have several connection factories referencing the same connection recovery entry. + * Because of that we need to make a count of the number of the instances that are referencing it, + * so we will remove it as soon as we are done */ + public synchronized int incrementUsage() + { + return ++usage; + } + + public synchronized int decrementUsage() + { + return --usage; + } + @Override public boolean equals(Object o) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; + if (this == o) + return true; + if (o == null || getClass() != o.getClass()) + return false; - HornetQResourceRecovery that = (HornetQResourceRecovery) o; + HornetQResourceRecovery that = (HornetQResourceRecovery)o; - if (config != null ? !config.equals(that.config) : that.config != null) return false; + if (config != null ? !config.equals(that.config) : that.config != null) + return false; return true; } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java 2012-03-09 21:09:17 UTC (rev 12281) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java 2012-03-09 21:11:33 UTC (rev 12282) @@ -22,6 +22,7 @@ import org.hornetq.api.core.HornetQException; import org.hornetq.api.core.client.ClientSession; import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.api.core.client.HornetQClient; import org.hornetq.api.core.client.ServerLocator; import org.hornetq.api.core.client.SessionFailureListener; import org.hornetq.core.logging.Logger; @@ -242,7 +243,7 @@ * @return the connectionFactory * @throws XAException for any problem */ - public XAResource getDelegate(boolean retry) throws XAException + private XAResource getDelegate(boolean retry) throws XAException { XAResource result = null; Exception error = null; @@ -316,7 +317,14 @@ try { - serverLocator = xaRecoveryConfig.getHornetQConnectionFactory().getServerLocator(); + if (xaRecoveryConfig.getDiscoveryConfiguration() != null) + { + serverLocator = HornetQClient.createServerLocator(xaRecoveryConfig.isHA(), xaRecoveryConfig.getDiscoveryConfiguration()); + } + else + { + serverLocator = HornetQClient.createServerLocator(xaRecoveryConfig.isHA(), xaRecoveryConfig.getTransportConfig()); + } serverLocator.disableFinalizeCheck(); csf = serverLocator.createSessionFactory(); if (xaRecoveryConfig.getUsername() == null) @@ -334,10 +342,29 @@ 1); } } - catch (HornetQException e) + catch (Throwable e) { + log.warn("Can't connect to " + xaRecoveryConfig + " on auto-generated resource recovery", e); + if (log.isDebugEnabled()) + { + log.debug(e.getMessage(), e); + } + + try + { + if (cs != null) cs.close(); + if (serverLocator != null) serverLocator.close(); + } + catch (Throwable ignored) + { + if (log.isTraceEnabled()) + { + log.trace(e.getMessage(), ignored); + } + } continue; } + cs.addFailureListener(this); synchronized (HornetQXAResourceWrapper.lock) @@ -392,9 +419,9 @@ oldServerLocator.close(); } } - catch (Exception ignored) + catch (Throwable ignored) { - HornetQXAResourceWrapper.log.trace("Ignored error during close", ignored); + HornetQXAResourceWrapper.log.debug("Ignored error during close", ignored); } } @@ -410,10 +437,9 @@ { log.warn(e.getMessage(), e); - if (e.errorCode == XAException.XA_RETRY) - { - close(); - } + + // If any exception happened, we close the connection so we may start fresh + close(); throw e; } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/RecoveryRegistry.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/RecoveryRegistry.java 2012-03-09 21:09:17 UTC (rev 12281) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/RecoveryRegistry.java 2012-03-09 21:11:33 UTC (rev 12282) @@ -28,7 +28,7 @@ */ public interface RecoveryRegistry { - void register(HornetQResourceRecovery resourceRecovery); + HornetQResourceRecovery register(HornetQResourceRecovery resourceRecovery); void unRegister(HornetQResourceRecovery xaRecoveryConfig); } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/XARecoveryConfig.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/XARecoveryConfig.java 2012-03-09 21:09:17 UTC (rev 12281) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/XARecoveryConfig.java 2012-03-09 21:11:33 UTC (rev 12282) @@ -13,10 +13,17 @@ package org.hornetq.jms.server.recovery; -import org.hornetq.jms.client.HornetQConnectionFactory; +import java.util.Arrays; +import org.hornetq.api.core.DiscoveryGroupConfiguration; +import org.hornetq.api.core.TransportConfiguration; + /** + * + * This represents the configuration of a single connection factory. + * * @author <a href="mailto:andy.taylor@jboss.com">Andy Taylor</a> + * @author Clebert Suconic * * A wrapper around info needed for the xa recovery resource * Date: 3/23/11 @@ -24,22 +31,46 @@ */ public class XARecoveryConfig { - private final HornetQConnectionFactory hornetQConnectionFactory; + + private final boolean ha; + private final TransportConfiguration[] transportConfiguration; + private final DiscoveryGroupConfiguration discoveryConfiguration; private final String username; private final String password; - public XARecoveryConfig(HornetQConnectionFactory hornetQConnectionFactory, String username, String password) + public XARecoveryConfig(final boolean ha, final TransportConfiguration[] transportConfiguration, final String username, final String password) { - this.hornetQConnectionFactory = hornetQConnectionFactory; + this.transportConfiguration = transportConfiguration; + this.discoveryConfiguration = null; this.username = username; this.password = password; + this.ha = ha; } - public HornetQConnectionFactory getHornetQConnectionFactory() + public XARecoveryConfig(final boolean ha, final DiscoveryGroupConfiguration discoveryConfiguration, final String username, final String password) { - return hornetQConnectionFactory; + this.discoveryConfiguration = discoveryConfiguration; + this.transportConfiguration = null; + this.username = username; + this.password = password; + this.ha = ha; } + + public boolean isHA() + { + return ha; + } + public DiscoveryGroupConfiguration getDiscoveryConfiguration() + { + return discoveryConfiguration; + } + + public TransportConfiguration[] getTransportConfig() + { + return transportConfiguration; + } + public String getUsername() { return username; @@ -49,24 +80,44 @@ { return password; } + + /* (non-Javadoc) + * @see java.lang.Object#hashCode() + */ + @Override + public int hashCode() + { + final int prime = 31; + int result = 1; + result = prime * result + ((discoveryConfiguration == null) ? 0 : discoveryConfiguration.hashCode()); + result = prime * result + Arrays.hashCode(transportConfiguration); + return result; + } + /* (non-Javadoc) + * @see java.lang.Object#equals(java.lang.Object) + */ @Override - public boolean equals(Object o) + public boolean equals(Object obj) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - - XARecoveryConfig that = (XARecoveryConfig) o; - - if (hornetQConnectionFactory != null ? !hornetQConnectionFactory.equals(that.hornetQConnectionFactory) : that.hornetQConnectionFactory != null) + if (this == obj) + return true; + if (obj == null) return false; - if (password != null ? !password.equals(that.password) : that.password != null) return false; - if (username != null ? !username.equals(that.username) : that.username != null) return false; - + if (getClass() != obj.getClass()) + return false; + XARecoveryConfig other = (XARecoveryConfig)obj; + if (discoveryConfiguration == null) + { + if (other.discoveryConfiguration != null) + return false; + } + else if (!discoveryConfiguration.equals(other.discoveryConfiguration)) + return false; + if (!Arrays.equals(transportConfiguration, other.transportConfiguration)) + return false; return true; } - - /* (non-Javadoc) * @see java.lang.Object#toString() @@ -74,20 +125,12 @@ @Override public String toString() { - return "XARecoveryConfig [hornetQConnectionFactory=" + hornetQConnectionFactory + + return "XARecoveryConfig [transportConfiguration = " + Arrays.toString(transportConfiguration) + + ", discoveryConfiguration = " + discoveryConfiguration + ", username=" + username + ", password=" + password + "]"; } - - @Override - public int hashCode() - { - int result = hornetQConnectionFactory != null ? hornetQConnectionFactory.hashCode() : 0; - result = 31 * result + (username != null ? username.hashCode() : 0); - result = 31 * result + (password != null ? password.hashCode() : 0); - return result; - } } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java 2012-03-09 21:09:17 UTC (rev 12281) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java 2012-03-09 21:11:33 UTC (rev 12282) @@ -21,21 +21,18 @@ */ package org.hornetq.ra.recovery; -import org.hornetq.api.core.DiscoveryGroupConfiguration; -import org.hornetq.api.core.TransportConfiguration; +import java.security.AccessController; +import java.security.PrivilegedAction; +import java.util.Set; + import org.hornetq.core.logging.Logger; import org.hornetq.jms.client.HornetQConnectionFactory; import org.hornetq.jms.server.recovery.HornetQResourceRecovery; import org.hornetq.jms.server.recovery.RecoveryRegistry; import org.hornetq.jms.server.recovery.XARecoveryConfig; -import org.hornetq.ra.Util; import org.hornetq.utils.ClassloadingUtil; +import org.hornetq.utils.ConcurrentHashSet; -import java.security.AccessController; -import java.security.PrivilegedAction; -import java.util.HashMap; -import java.util.Map; - /** * @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a> * 9/21/11 @@ -47,9 +44,9 @@ private RecoveryRegistry registry; private String resourceRecoveryClassNames = "org.jboss.as.messaging.jms.AS7RecoveryRegistry;org.jboss.as.integration.hornetq.recovery.AS5RecoveryRegistry"; + + private final Set<HornetQResourceRecovery> resources = new ConcurrentHashSet<HornetQResourceRecovery>(); - private Map<XARecoveryConfig, HornetQResourceRecovery> configMap = new HashMap<XARecoveryConfig, HornetQResourceRecovery>(); - public void start() { locateRecoveryRegistry(); @@ -59,17 +56,45 @@ { log.debug("registering recovery for factory : " + factory); - if(!isRegistered(factory) && registry != null) + HornetQResourceRecovery resourceRecovery = newResourceRecovery(factory, userName, password); + + if (registry != null) { - XARecoveryConfig xaRecoveryConfig = new XARecoveryConfig(factory, userName, password); - HornetQResourceRecovery resourceRecovery = new HornetQResourceRecovery(xaRecoveryConfig); - registry.register(resourceRecovery); - configMap.put(xaRecoveryConfig, resourceRecovery); - return resourceRecovery; + resourceRecovery = registry.register(resourceRecovery); + if (resourceRecovery != null) + { + resources.add(resourceRecovery); + } } - return null; + + return resourceRecovery; } + /** + * @param factory + * @param userName + * @param password + * @return + */ + private HornetQResourceRecovery newResourceRecovery(HornetQConnectionFactory factory, + String userName, + String password) + { + XARecoveryConfig xaRecoveryConfig; + + if (factory.getServerLocator().getDiscoveryGroupConfiguration() != null) + { + xaRecoveryConfig = new XARecoveryConfig(factory.getServerLocator().isHA(), factory.getServerLocator().getDiscoveryGroupConfiguration(), userName, password); + } + else + { + xaRecoveryConfig = new XARecoveryConfig(factory.getServerLocator().isHA(), factory.getServerLocator().getStaticTransportConfigurations(), userName, password); + } + + HornetQResourceRecovery resourceRecovery = new HornetQResourceRecovery(xaRecoveryConfig); + return resourceRecovery; + } + public void unRegister(HornetQResourceRecovery resourceRecovery) { registry.unRegister(resourceRecovery); @@ -77,11 +102,11 @@ public void stop() { - for (HornetQResourceRecovery hornetQResourceRecovery : configMap.values()) + for (HornetQResourceRecovery hornetQResourceRecovery : resources) { registry.unRegister(hornetQResourceRecovery); } - configMap.clear(); + resources.clear(); } private void locateRecoveryRegistry() @@ -108,9 +133,9 @@ { registry = new RecoveryRegistry() { - public void register(HornetQResourceRecovery resourceRecovery) + public HornetQResourceRecovery register(HornetQResourceRecovery resourceRecovery) { - //no op + return null; } public void unRegister(HornetQResourceRecovery xaRecoveryConfig) @@ -125,52 +150,6 @@ } } - - public boolean isRegistered(HornetQConnectionFactory factory) - { - for (XARecoveryConfig xaRecoveryConfig : configMap.keySet()) - { - TransportConfiguration[] transportConfigurations = factory.getServerLocator().getStaticTransportConfigurations(); - - if (transportConfigurations != null) - { - TransportConfiguration[] xaConfigurations = xaRecoveryConfig.getHornetQConnectionFactory().getServerLocator().getStaticTransportConfigurations(); - if(xaConfigurations == null) - { - break; - } - if(transportConfigurations.length != xaConfigurations.length) - { - break; - } - boolean theSame=true; - for(int i = 0; i < transportConfigurations.length; i++) - { - TransportConfiguration tc = transportConfigurations[i]; - TransportConfiguration xaTc = xaConfigurations[i]; - if(!tc.equals(xaTc)) - { - theSame = false; - break; - } - } - if(theSame) - { - return theSame; - } - } - else - { - DiscoveryGroupConfiguration discoveryGroupConfiguration = xaRecoveryConfig.getHornetQConnectionFactory().getServerLocator().getDiscoveryGroupConfiguration(); - if(discoveryGroupConfiguration != null && discoveryGroupConfiguration.equals(factory.getDiscoveryGroupConfiguration())) - { - return true; - } - } - } - return false; - } - /** This seems duplicate code all over the place, but for security reasons we can't let something like this to be open in a * utility class, as it would be a door to load anything you like in a safe VM. * For that reason any class trying to do a privileged block should do with the AccessController directly.