[hornetq-commits] JBoss hornetq SVN: r12282 - in branches/Branch_2_2_AS7/src/main/org/hornetq: jms/server/recovery and 1 other directories.
do-not-reply at jboss.org
do-not-reply at jboss.org
Fri Mar 9 16:11:34 EST 2012
Author: clebert.suconic at 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 at 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 at 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.
More information about the hornetq-commits
mailing list