This closes #127 lock between data tools and running servers
This commit is contained in:
commit
1d407c2c3e
|
@ -33,12 +33,14 @@ public final class CompactJournal extends DataAbstract implements Action {
|
|||
public Object execute(ActionContext context) throws Exception {
|
||||
super.execute(context);
|
||||
try {
|
||||
testLock();
|
||||
Configuration configuration = getFileConfiguration();
|
||||
compactJournal(new File(getJournal()), "activemq-data", "amq", configuration.getJournalMinFiles(), configuration.getJournalFileSize(), null);
|
||||
compactJournal(new File(getBinding()), "activemq-bindings", "bindings", 2, 1048576, null);
|
||||
}
|
||||
catch (Exception e) {
|
||||
treatError(e, "data", "compact");
|
||||
return e;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
|
|
@ -18,9 +18,15 @@
|
|||
package org.apache.activemq.artemis.cli.commands.tools;
|
||||
|
||||
import java.io.File;
|
||||
import java.nio.channels.FileLock;
|
||||
|
||||
import io.airlift.airline.Option;
|
||||
import org.apache.activemq.artemis.cli.commands.Configurable;
|
||||
import org.apache.activemq.artemis.core.config.Configuration;
|
||||
import org.apache.activemq.artemis.core.server.JournalType;
|
||||
import org.apache.activemq.artemis.core.server.impl.AIOFileLockNodeManager;
|
||||
import org.apache.activemq.artemis.core.server.impl.FileLockNodeManager;
|
||||
import org.apache.activemq.artemis.jlibaio.LibaioContext;
|
||||
|
||||
/**
|
||||
* Abstract class for places where you need bindings, journal paging and large messages configuration
|
||||
|
@ -39,6 +45,31 @@ public abstract class DataAbstract extends Configurable {
|
|||
@Option(name = "--large-messages", description = "The folder used for large-messages (default from broker.xml)")
|
||||
public String largeMessges;
|
||||
|
||||
|
||||
protected void testLock() throws Exception {
|
||||
|
||||
FileLockNodeManager fileLockNodeManager;
|
||||
Configuration configuration = getFileConfiguration();
|
||||
if (getFileConfiguration().getJournalType() == JournalType.ASYNCIO && LibaioContext.isLoaded()) {
|
||||
fileLockNodeManager = new AIOFileLockNodeManager(new File(getJournal()), false, configuration.getJournalLockAcquisitionTimeout());
|
||||
}
|
||||
else {
|
||||
fileLockNodeManager = new FileLockNodeManager(new File(getJournal()), false, configuration.getJournalLockAcquisitionTimeout());
|
||||
}
|
||||
|
||||
fileLockNodeManager.start();
|
||||
|
||||
try (FileLock lock = fileLockNodeManager.tryLockLive()) {
|
||||
if (lock == null) {
|
||||
throw new RuntimeException("Server is locked!");
|
||||
}
|
||||
}
|
||||
finally {
|
||||
fileLockNodeManager.stop();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public String getLargeMessages() throws Exception {
|
||||
if (largeMessges == null) {
|
||||
largeMessges = getFileConfiguration().getLargeMessagesLocation().getAbsolutePath();
|
||||
|
|
|
@ -63,10 +63,12 @@ public class PrintData extends DataAbstract implements Action {
|
|||
public Object execute(ActionContext context) throws Exception {
|
||||
super.execute(context);
|
||||
try {
|
||||
testLock();
|
||||
printData(new File(getBinding()), new File(getJournal()), new File(getPaging()));
|
||||
}
|
||||
catch (Exception e) {
|
||||
treatError(e, "data", "print");
|
||||
return e;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
|
|
@ -129,10 +129,12 @@ public final class XmlDataExporter extends DataAbstract implements Action {
|
|||
super.execute(context);
|
||||
|
||||
try {
|
||||
testLock();
|
||||
process(context.out, getBinding(), getJournal(), getPaging(), getLargeMessages());
|
||||
}
|
||||
catch (Exception e) {
|
||||
treatError(e, "data", "exp");
|
||||
return e;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
|
|
@ -93,6 +93,15 @@ public class ArtemisTest {
|
|||
// Some exceptions may happen on the initialization, but they should be ok on start the basic core protocol
|
||||
Artemis.execute("run");
|
||||
|
||||
Object object = Artemis.execute("data", "print");
|
||||
Assert.assertTrue("An error was expected", object != null && object instanceof Throwable);
|
||||
|
||||
object = Artemis.execute("data", "compact");
|
||||
Assert.assertTrue("An error was expected", object != null && object instanceof Throwable);
|
||||
|
||||
object = Artemis.execute("data", "exp");
|
||||
Assert.assertTrue("An error was expected", object != null && object instanceof Throwable);
|
||||
|
||||
try (ServerLocator locator = ServerLocatorImpl.newLocator("tcp://localhost:61616");
|
||||
ClientSessionFactory factory = locator.createSessionFactory();
|
||||
ClientSession coreSession = factory.createSession()) {
|
||||
|
|
|
@ -21,6 +21,7 @@ import java.io.IOException;
|
|||
import java.io.RandomAccessFile;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.FileChannel;
|
||||
import java.nio.channels.FileLock;
|
||||
|
||||
import org.apache.activemq.artemis.api.core.ActiveMQIllegalStateException;
|
||||
import org.apache.activemq.artemis.api.core.SimpleString;
|
||||
|
@ -130,6 +131,8 @@ public abstract class NodeManager implements ActiveMQComponent {
|
|||
releaseBackup();
|
||||
}
|
||||
|
||||
public abstract FileLock tryLockLive();
|
||||
|
||||
/**
|
||||
* Ensures existence of persistent information about the server's nodeID.
|
||||
* <p>
|
||||
|
|
|
@ -347,9 +347,10 @@ public class ActiveMQServerImpl implements ActiveMQServer {
|
|||
protected NodeManager createNodeManager(final File directory, boolean replicatingBackup) {
|
||||
NodeManager manager;
|
||||
if (!configuration.isPersistenceEnabled()) {
|
||||
manager = new InVMNodeManager(replicatingBackup);
|
||||
return new InVMNodeManager(replicatingBackup);
|
||||
}
|
||||
else if (configuration.getJournalType() == JournalType.ASYNCIO && LibaioContext.isLoaded()) {
|
||||
|
||||
if (configuration.getJournalType() == JournalType.ASYNCIO && LibaioContext.isLoaded()) {
|
||||
manager = new AIOFileLockNodeManager(directory, replicatingBackup, configuration.getJournalLockAcquisitionTimeout());
|
||||
}
|
||||
else {
|
||||
|
|
|
@ -73,6 +73,15 @@ public class FileLockNodeManager extends NodeManager {
|
|||
super.start();
|
||||
}
|
||||
|
||||
public FileLock tryLockLive() {
|
||||
try {
|
||||
return tryLock(FileLockNodeManager.LIVE_LOCK_POS);
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new RuntimeException(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isAwaitingFailback() throws Exception {
|
||||
return getState() == FileLockNodeManager.FAILINGBACK;
|
||||
|
|
|
@ -18,6 +18,7 @@ package org.apache.activemq.artemis.core.server.impl;
|
|||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.nio.channels.FileLock;
|
||||
import java.util.concurrent.Semaphore;
|
||||
|
||||
import org.apache.activemq.artemis.api.core.ActiveMQIllegalStateException;
|
||||
|
@ -95,6 +96,12 @@ public final class InVMNodeManager extends NodeManager {
|
|||
backupLock.acquire();
|
||||
}
|
||||
|
||||
@Override
|
||||
public FileLock tryLockLive() {
|
||||
// no op.. doesn't make sense on InVM
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void startLiveNode() throws Exception {
|
||||
state = FAILING_BACK;
|
||||
|
|
|
@ -16,6 +16,11 @@
|
|||
*/
|
||||
package org.apache.activemq.artemis.core.server.impl;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
|
||||
import org.apache.activemq.artemis.api.core.ActiveMQException;
|
||||
import org.apache.activemq.artemis.api.core.Pair;
|
||||
import org.apache.activemq.artemis.api.core.SimpleString;
|
||||
|
@ -32,11 +37,6 @@ import org.apache.activemq.artemis.core.server.cluster.ActiveMQServerSideProtoco
|
|||
import org.apache.activemq.artemis.core.server.cluster.ha.LiveOnlyPolicy;
|
||||
import org.apache.activemq.artemis.core.server.cluster.ha.ScaleDownPolicy;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
|
||||
public class LiveOnlyActivation extends Activation {
|
||||
|
||||
//this is how we act when we initially start as live
|
||||
|
@ -55,6 +55,11 @@ public class LiveOnlyActivation extends Activation {
|
|||
|
||||
public void run() {
|
||||
try {
|
||||
|
||||
/* We will hold a lock here so print-data and other tools
|
||||
* won't be able to run */
|
||||
activeMQServer.getNodeManager().startLiveNode();
|
||||
|
||||
activeMQServer.initialisePart1(false);
|
||||
|
||||
activeMQServer.initialisePart2(false);
|
||||
|
|
|
@ -17,6 +17,7 @@
|
|||
package org.apache.activemq.artemis.tests.integration.discovery;
|
||||
|
||||
import java.net.InetAddress;
|
||||
import java.nio.channels.FileLock;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.Comparator;
|
||||
|
@ -176,6 +177,11 @@ public class DiscoveryBaseTest extends ActiveMQTestBase {
|
|||
this.setNodeID(nodeID);
|
||||
}
|
||||
|
||||
@Override
|
||||
public FileLock tryLockLive() {
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void awaitLiveNode() throws Exception {
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue