[hornetq-commits] JBoss hornetq SVN: r11409 - branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl.
do-not-reply at jboss.org
do-not-reply at jboss.org
Sat Sep 24 16:49:52 EDT 2011
Author: clebert.suconic at jboss.com
Date: 2011-09-24 16:49:52 -0400 (Sat, 24 Sep 2011)
New Revision: 11409
Modified:
branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
Log:
fixing test
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-09-24 16:16:41 UTC (rev 11408)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-09-24 20:49:52 UTC (rev 11409)
@@ -65,11 +65,13 @@
import org.hornetq.core.paging.impl.PagingManagerImpl;
import org.hornetq.core.paging.impl.PagingStoreFactoryNIO;
import org.hornetq.core.persistence.GroupingInfo;
+import org.hornetq.core.persistence.OperationContext;
import org.hornetq.core.persistence.QueueBindingInfo;
import org.hornetq.core.persistence.StorageManager;
import org.hornetq.core.persistence.config.PersistedAddressSetting;
import org.hornetq.core.persistence.config.PersistedRoles;
import org.hornetq.core.persistence.impl.journal.JournalStorageManager;
+import org.hornetq.core.persistence.impl.journal.OperationContextImpl;
import org.hornetq.core.persistence.impl.nullpm.NullStorageManager;
import org.hornetq.core.postoffice.Binding;
import org.hornetq.core.postoffice.DuplicateIDCache;
@@ -137,11 +139,13 @@
// ------------------------------------------------------------------------------------
private static final Logger log = Logger.getLogger(HornetQServerImpl.class);
-
- // JMS Topics (which are outside of the scope of the core API) will require a dumb subscription with a dummy-filter at this current version
+
+ // JMS Topics (which are outside of the scope of the core API) will require a dumb subscription with a dummy-filter
+ // at this current version
// as a way to keep its existence valid and TCK tests
// That subscription needs an invalid filter, however paging needs to ignore any subscription with this filter.
- // For that reason, this filter needs to be rejected on paging or any other component on the system, and just be ignored for any purpose
+ // For that reason, this filter needs to be rejected on paging or any other component on the system, and just be
+ // ignored for any purpose
// It's declared here as this filter is considered a global ignore
public static final String GENERIC_IGNORED_FILTER = "__HQX=-1";
@@ -151,7 +155,6 @@
// Attributes
// -----------------------------------------------------------------------------------
-
private final Version version;
private final HornetQSecurityManager securityManager;
@@ -161,7 +164,7 @@
private final MBeanServer mbeanServer;
private volatile boolean started;
-
+
private volatile boolean stopped;
private volatile SecurityStore securityStore;
@@ -171,7 +174,7 @@
private volatile QueueFactory queueFactory;
private volatile PagingManager pagingManager;
-
+
private volatile PostOffice postOffice;
private volatile ExecutorService threadPool;
@@ -193,7 +196,7 @@
private volatile RemotingService remotingService;
private volatile ManagementService managementService;
-
+
private volatile ConnectorsService connectorsService;
private MemoryManager memoryManager;
@@ -214,7 +217,7 @@
private boolean initialised;
- // private FailoverManager replicationFailoverManager;
+ // private FailoverManager replicationFailoverManager;
private ReplicationManager replicationManager;
@@ -223,17 +226,16 @@
private final Set<ActivateCallback> activateCallbacks = new HashSet<ActivateCallback>();
private volatile GroupingHandler groupingHandler;
-
+
private NodeManager nodeManager;
-
+
// Used to identify the server on tests... useful on debugging testcases
private String identity;
-
+
private Thread backupActivationThread;
private Activation activation;
-
// Constructors
// ---------------------------------------------------------------------------------
@@ -312,74 +314,89 @@
public synchronized void start() throws Exception
{
- stopped = false;
-
- initialiseLogging();
+ OperationContextImpl.clearContext();
- checkJournalDirectory();
+ try
+ {
+ stopped = false;
- nodeManager = createNodeManager(configuration.getJournalDirectory());
+ initialiseLogging();
- nodeManager.start();
+ checkJournalDirectory();
- if (started)
- {
- HornetQServerImpl.log.info((configuration.isBackup() ? "backup" : "live") + " is already started, ignoring the call to start..");
- return;
- }
+ nodeManager = createNodeManager(configuration.getJournalDirectory());
- HornetQServerImpl.log.info((configuration.isBackup() ? "backup" : "live") + " server is starting with configuration " + configuration);
+ nodeManager.start();
- if (configuration.isRunSyncSpeedTest())
- {
- SyncSpeedTest test = new SyncSpeedTest();
-
- test.run();
- }
-
- if (!configuration.isBackup())
- {
- if (configuration.isSharedStore() && configuration.isPersistenceEnabled())
+ if (started)
{
- activation = new SharedStoreLiveActivation();
+ HornetQServerImpl.log.info((configuration.isBackup() ? "backup" : "live") + " is already started, ignoring the call to start..");
+ return;
+ }
- // This should block until the lock is got
+ HornetQServerImpl.log.info((configuration.isBackup() ? "backup" : "live") + " server is starting with configuration " +
+ configuration);
- activation.run();
- }
- else
+ if (configuration.isRunSyncSpeedTest())
{
- activation = new NoSharedStoreLiveActivation();
+ SyncSpeedTest test = new SyncSpeedTest();
- activation.run();
+ test.run();
}
- started = true;
- HornetQServerImpl.log.info("HornetQ Server version " + getVersion().getFullVersion() + " [" + nodeManager.getNodeId() + "]" + (this.identity != null ? " (" + identity : ")") + " started");
- }
+ if (!configuration.isBackup())
+ {
+ if (configuration.isSharedStore() && configuration.isPersistenceEnabled())
+ {
+ activation = new SharedStoreLiveActivation();
+ // This should block until the lock is got
- // The activation on fail-back may change the value of isBackup, for that reason we are not using else here
- if (configuration.isBackup())
- {
- if (configuration.isSharedStore())
- {
- activation = new SharedStoreBackupActivation();
+ activation.run();
+ }
+ else
+ {
+ activation = new NoSharedStoreLiveActivation();
+
+ activation.run();
+ }
+ started = true;
+
+ HornetQServerImpl.log.info("HornetQ Server version " + getVersion().getFullVersion() +
+ " [" +
+ nodeManager.getNodeId() +
+ "]" +
+ (this.identity != null ? " (" + identity : ")") +
+ " started");
}
- else
+
+ // The activation on fail-back may change the value of isBackup, for that reason we are not using else here
+ if (configuration.isBackup())
{
- // Replicated
+ if (configuration.isSharedStore())
+ {
+ activation = new SharedStoreBackupActivation();
+ }
+ else
+ {
+ // Replicated
- activation = new SharedNothingBackupActivation();
+ activation = new SharedNothingBackupActivation();
+ }
+
+ backupActivationThread = new Thread(activation, "Activation for server " + this);
+ backupActivationThread.start();
}
- backupActivationThread = new Thread(activation, "Activation for server " + this);
- backupActivationThread.start();
+ // start connector service
+ connectorsService = new ConnectorsService(configuration, storageManager, scheduledPool, postOffice);
+ connectorsService.start();
}
-
- // start connector service
- connectorsService = new ConnectorsService(configuration, storageManager, scheduledPool, postOffice);
- connectorsService.start();
+ finally
+ {
+ // this avoids embedded applications using dirty contexts from startup
+ OperationContextImpl.clearContext();
+ }
}
@Override
@@ -400,7 +417,7 @@
stopped = true;
stop(configuration.isFailoverOnServerShutdown());
}
-
+
public void threadDump(final String reason)
{
StringWriter str = new StringWriter();
@@ -445,20 +462,21 @@
}
connectorsService.stop();
- //we stop the groupinghandler before we stop te cluster manager so binding mappings aren't removed in case of failover
+ // we stop the groupinghandler before we stop te cluster manager so binding mappings aren't removed in case of
+ // failover
if (groupingHandler != null)
{
managementService.removeNotificationListener(groupingHandler);
groupingHandler = null;
}
-
+
if (clusterManager != null)
{
clusterManager.stop();
}
}
-
+
// We close all the exception in an attempt to let any pending IO to finish
// to avoid scenarios where the send or ACK got to disk but the response didn't get to the client
// It may still be possible to have this scenario on a real failure (without the use of XA)
@@ -545,9 +563,9 @@
{
memoryManager.stop();
}
-
+
threadPool.shutdown();
-
+
scheduledPool.shutdown();
try
@@ -563,7 +581,6 @@
}
threadPool = null;
-
try
{
if (!scheduledPool.awaitTermination(10, TimeUnit.SECONDS))
@@ -577,7 +594,7 @@
}
threadPool = null;
-
+
scheduledPool = null;
pagingManager = null;
@@ -610,17 +627,20 @@
nodeManager.stop();
nodeManager = null;
-
+
addressSettingsRepository.clearListeners();
-
+
addressSettingsRepository.clearCache();
- HornetQServerImpl.log.info("HornetQ Server version " + getVersion().getFullVersion() + " [" + tempNodeID + "] stopped");
+ HornetQServerImpl.log.info("HornetQ Server version " + getVersion().getFullVersion() +
+ " [" +
+ tempNodeID +
+ "] stopped");
Logger.reset();
}
- }
+ }
// HornetQServer implementation
// -----------------------------------------------------------
@@ -629,33 +649,33 @@
{
StringWriter str = new StringWriter();
PrintWriter out = new PrintWriter(str);
-
+
out.println("Information about server " + this.identity);
out.println("Cluster Connection:" + this.getClusterManager().describe());
-
+
return str.toString();
}
-
+
public void setIdentity(String identity)
{
this.identity = identity;
}
-
+
public String getIdentity()
{
return identity;
}
-
+
public ScheduledExecutorService getScheduledPool()
{
return scheduledPool;
}
-
+
public ExecutorService getThreadPool()
{
return threadPool;
}
-
+
public Configuration getConfiguration()
{
return configuration;
@@ -665,7 +685,7 @@
{
return mbeanServer;
}
-
+
public PagingManager getPagingManager()
{
return pagingManager;
@@ -695,7 +715,7 @@
{
return securityRepository;
}
-
+
public NodeManager getNodeManager()
{
return nodeManager;
@@ -802,7 +822,7 @@
{
// getSessions is called here in a try to minimize locking the Server while this check is being done
Set<ServerSession> allSessions = getSessions();
-
+
for (ServerSession session : allSessions)
{
String metaValue = session.getMetaData(key);
@@ -811,10 +831,10 @@
return true;
}
}
-
+
return false;
}
-
+
public synchronized List<ServerSession> getSessions(final String connectionID)
{
Set<Entry<String, ServerSession>> sessionEntries = sessions.entrySet();
@@ -865,7 +885,7 @@
public SimpleString getNodeID()
{
- return nodeManager == null?null:nodeManager.getNodeId();
+ return nodeManager == null ? null : nodeManager.getNodeId();
}
public Queue createQueue(final SimpleString address,
@@ -876,24 +896,24 @@
{
return createQueue(address, queueName, filterString, durable, temporary, false);
}
-
+
public Queue locateQueue(SimpleString queueName) throws Exception
{
Binding binding = postOffice.getBinding(queueName);
-
+
if (binding == null)
{
return null;
}
-
+
Bindable queue = binding.getBindable();
-
+
if (!(queue instanceof Queue))
{
throw new IllegalStateException("locateQueue should only be used to locate queues");
}
-
- return (Queue) binding.getBindable();
+
+ return (Queue)binding.getBindable();
}
public Queue deployQueue(final SimpleString address,
@@ -919,7 +939,7 @@
}
Queue queue = (Queue)binding.getBindable();
-
+
if (queue.getPageSubscription() != null)
{
queue.getPageSubscription().close();
@@ -997,7 +1017,6 @@
return connectorsService;
}
-
public synchronized boolean checkActivate() throws Exception
{
if (configuration.isBackup())
@@ -1081,7 +1100,7 @@
managementService.registerDivert(divert, config);
}
-
+
public void destroyDivert(SimpleString name) throws Exception
{
Binding binding = postOffice.getBinding(name);
@@ -1097,8 +1116,6 @@
postOffice.removeBinding(name);
}
-
-
public void deployBridge(BridgeConfiguration config) throws Exception
{
if (clusterManager != null)
@@ -1106,7 +1123,7 @@
clusterManager.deployBridge(config, true);
}
}
-
+
public void destroyBridge(String name) throws Exception
{
if (clusterManager != null)
@@ -1114,14 +1131,14 @@
clusterManager.destroyBridge(name);
}
}
-
+
public ServerSession getSessionByID(String sessionName)
{
return sessions.get(sessionName);
}
-
+
// PUBLIC -------
-
+
public String toString()
{
if (identity != null)
@@ -1134,8 +1151,6 @@
}
}
-
-
// Package protected
// ----------------------------------------------------------------------------
@@ -1146,30 +1161,30 @@
* Protected so tests can change this behaviour
* @param backupConnector
*/
-// protected FailoverManagerImpl createBackupConnectionFailoverManager(final TransportConfiguration backupConnector,
-// final ExecutorService threadPool,
-// final ScheduledExecutorService scheduledPool)
-// {
-// return new FailoverManagerImpl((ClientSessionFactory)null,
-// backupConnector,
-// null,
-// false,
-// HornetQClient.DEFAULT_CALL_TIMEOUT,
-// HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
-// HornetQClient.DEFAULT_CONNECTION_TTL,
-// 0,
-// 1.0d,
-// 0,
-// 1,
-// false,
-// threadPool,
-// scheduledPool,
-// null);
-// }
+ // protected FailoverManagerImpl createBackupConnectionFailoverManager(final TransportConfiguration backupConnector,
+ // final ExecutorService threadPool,
+ // final ScheduledExecutorService scheduledPool)
+ // {
+ // return new FailoverManagerImpl((ClientSessionFactory)null,
+ // backupConnector,
+ // null,
+ // false,
+ // HornetQClient.DEFAULT_CALL_TIMEOUT,
+ // HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+ // HornetQClient.DEFAULT_CONNECTION_TTL,
+ // 0,
+ // 1.0d,
+ // 0,
+ // 1,
+ // false,
+ // threadPool,
+ // scheduledPool,
+ // null);
+ // }
protected PagingManager createPagingManager()
{
-
+
return new PagingManagerImpl(new PagingStoreFactoryNIO(configuration.getPagingDirectory(),
(long)configuration.getJournalBufferSize_NIO(),
scheduledPool,
@@ -1313,11 +1328,9 @@
nodeManager.getUUID(),
configuration.isBackup(),
configuration.isClustered());
-
clusterManager.deploy();
-
remotingService = new RemotingServiceImpl(clusterManager, configuration, this, managementService, scheduledPool);
messagingServerControl = managementService.registerServer(postOffice,
@@ -1341,7 +1354,7 @@
addressSettingsDeployer.start();
}
-
+
deployAddressSettingsFromConfiguration();
storageManager.start();
@@ -1385,15 +1398,14 @@
private void initialisePart2() throws Exception
{
// Load the journal and populate queues, transactions and caches in memory
-
-
+
if (stopped)
{
return;
}
-
+
pagingManager.reloadStores();
-
+
JournalLoadInformation[] journalInfo = loadJournals();
compareJournals(journalInfo);
@@ -1507,13 +1519,17 @@
for (QueueBindingInfo queueBindingInfo : queueBindingInfos)
{
queueBindingInfosMap.put(queueBindingInfo.getId(), queueBindingInfo);
-
- if (queueBindingInfo.getFilterString() == null || !queueBindingInfo.getFilterString().toString().equals(GENERIC_IGNORED_FILTER))
+
+ if (queueBindingInfo.getFilterString() == null || !queueBindingInfo.getFilterString()
+ .toString()
+ .equals(GENERIC_IGNORED_FILTER))
{
Filter filter = FilterImpl.createFilter(queueBindingInfo.getFilterString());
-
- PageSubscription subscription = pagingManager.getPageStore(queueBindingInfo.getAddress()).getCursorProvier().createSubscription(queueBindingInfo.getId(), filter, true);
-
+
+ PageSubscription subscription = pagingManager.getPageStore(queueBindingInfo.getAddress())
+ .getCursorProvier()
+ .createSubscription(queueBindingInfo.getId(), filter, true);
+
Queue queue = queueFactory.createQueue(queueBindingInfo.getId(),
queueBindingInfo.getAddress(),
queueBindingInfo.getQueueName(),
@@ -1521,18 +1537,17 @@
subscription,
true,
false);
-
+
Binding binding = new LocalQueueBinding(queueBindingInfo.getAddress(), queue, nodeManager.getNodeId());
-
+
queues.put(queueBindingInfo.getId(), queue);
-
+
postOffice.addBinding(binding);
-
+
managementService.registerAddress(queueBindingInfo.getAddress());
managementService.registerQueue(queue, queueBindingInfo.getAddress(), storageManager);
}
-
-
+
}
for (GroupingInfo groupingInfo : groupingInfos)
@@ -1546,7 +1561,7 @@
}
Map<SimpleString, List<Pair<byte[], Long>>> duplicateIDMap = new HashMap<SimpleString, List<Pair<byte[], Long>>>();
-
+
HashSet<Pair<Long, Long>> pendingLargeMessages = new HashSet<Pair<Long, Long>>();
journalInfo[1] = storageManager.loadMessageJournal(postOffice,
@@ -1568,7 +1583,7 @@
cache.load(entry.getValue());
}
}
-
+
for (Pair<Long, Long> msgToDelete : pendingLargeMessages)
{
log.info("Deleting pending large message as it wasn't completed:" + msgToDelete);
@@ -1631,19 +1646,20 @@
}
Filter filter = FilterImpl.createFilter(filterString);
-
+
long queueID = storageManager.generateUniqueID();
PageSubscription pageSubscription;
-
-
+
if (filterString != null && filterString.toString().equals(GENERIC_IGNORED_FILTER))
{
pageSubscription = null;
}
else
{
- pageSubscription = pagingManager.getPageStore(address).getCursorProvier().createSubscription(queueID, filter, durable);
+ pageSubscription = pagingManager.getPageStore(address)
+ .getCursorProvier()
+ .createSubscription(queueID, filter, durable);
}
final Queue queue = queueFactory.createQueue(queueID,
@@ -1759,32 +1775,31 @@
}
else
{
- throw new IllegalArgumentException("Directory " + journalDir +
- " does not exist and will not be created");
+ throw new IllegalArgumentException("Directory " + journalDir + " does not exist and will not be created");
}
}
}
-
+
/**
* To be called by backup trying to fail back the server
*/
private void startFailbackChecker()
{
- scheduledPool.scheduleAtFixedRate(new FailbackChecker(), 1000l, 1000l, TimeUnit.MILLISECONDS);
+ scheduledPool.scheduleAtFixedRate(new FailbackChecker(), 1000l, 1000l, TimeUnit.MILLISECONDS);
}
-
// Inner classes
// --------------------------------------------------------------------------------
-
+
class FailbackChecker implements Runnable
{
boolean restarting = false;
+
public void run()
{
try
{
- if(!restarting && nodeManager.isAwaitingFailback())
+ if (!restarting && nodeManager.isAwaitingFailback())
{
log.info("live server wants to restart, restarting server in backup");
restarting = true;
@@ -1819,8 +1834,6 @@
}
}
-
-
private class SharedStoreLiveActivation implements Activation
{
public void run()
@@ -1830,7 +1843,7 @@
log.info("Waiting to obtain live lock");
checkJournalDirectory();
-
+
if (log.isDebugEnabled())
{
log.debug("First part initialization on " + this);
@@ -1838,9 +1851,10 @@
initialisePart1();
- if(nodeManager.isBackupLive())
+ if (nodeManager.isBackupLive())
{
- //looks like we've failed over at some point need to inform that we are the backup so when the current live
+ // looks like we've failed over at some point need to inform that we are the backup so when the current
+ // live
// goes down they failover to us
if (log.isDebugEnabled())
{
@@ -1857,9 +1871,9 @@
{
return;
}
-
+
initialisePart2();
-
+
log.info("Server is now live");
}
catch (Exception e)
@@ -1870,7 +1884,7 @@
public void close(boolean permanently) throws Exception
{
- if(permanently)
+ if (permanently)
{
nodeManager.crashLiveServer();
}
@@ -1881,7 +1895,6 @@
}
}
-
private class SharedStoreBackupActivation implements Activation
{
public void run()
@@ -1891,50 +1904,53 @@
nodeManager.startBackup();
initialisePart1();
-
+
clusterManager.start();
started = true;
- log.info("HornetQ Backup Server version " + getVersion().getFullVersion() + " [" + nodeManager.getNodeId() + "] started, waiting live to fail before it gets active");
+ log.info("HornetQ Backup Server version " + getVersion().getFullVersion() +
+ " [" +
+ nodeManager.getNodeId() +
+ "] started, waiting live to fail before it gets active");
nodeManager.awaitLiveNode();
-
+
configuration.setBackup(false);
-
+
if (stopped)
{
return;
}
-
+
initialisePart2();
-
+
clusterManager.activate();
log.info("Backup Server is now live");
nodeManager.releaseBackup();
- if(configuration.isAllowAutoFailBack())
+ if (configuration.isAllowAutoFailBack())
{
startFailbackChecker();
}
}
catch (InterruptedException e)
{
- //this is ok, we are being stopped
+ // this is ok, we are being stopped
}
catch (ClosedChannelException e)
{
- //this is ok too, we are being stopped
+ // this is ok too, we are being stopped
}
catch (Exception e)
{
- if(!(e.getCause() instanceof InterruptedException))
+ if (!(e.getCause() instanceof InterruptedException))
{
log.error("Failure in initialisation", e);
}
}
- catch(Throwable e)
+ catch (Throwable e)
{
log.error("Failure in initialisation", e);
}
@@ -1956,7 +1972,7 @@
nodeManager.interrupt();
backupActivationThread.interrupt();
-
+
backupActivationThread.join(1000);
}
@@ -1970,10 +1986,10 @@
}
else
{
- //if we are now live, behave as live
+ // if we are now live, behave as live
// We need to delete the file too, otherwise the backup will failover when we shutdown or if the backup is
// started before the live
- if(permanently)
+ if (permanently)
{
nodeManager.crashLiveServer();
}
@@ -1984,7 +2000,7 @@
}
}
}
-
+
private interface Activation extends Runnable
{
void close(boolean permanently) throws Exception;
@@ -2044,5 +2060,4 @@
}
}
-
}
More information about the hornetq-commits
mailing list