ARTEMIS-4559 refactor HA docs & code/module naming

This commit does the following:

 - Updates HA docs including the chapter on network isolation (i.e.
   split brain). The network isolation chapter is now more about
   high-level explanation and the HA doc now has all the configuration
   parameters.
 - Changes references to "pluggable quorum voting" to "pluggable lock
   manager." The pluggable functionality really isn't about voting.
   Conceptually is much more like the functionality you'd get from a
   distributed lock so this naming is more clear. Both the docs and the
   code have been changed.
 - Reorganize lock manager modules as sub-modules. The API and RI
   modules are renamed, but that should be OK based on the
   "experimental" tag that's been on this feature up to this point.
 - Remove the "experimental" tag from the lock manager.

These changes will not break folks using the standalone broker. However,
they will break folks embedding the broker *if* they are using the
artemis-quorum-ri or artemis-quorum-api modules or the
o.a.a.a.c.c.h.DistributedPrimitiveManagerConfiguration class.

There are no functional changes here. Renaming these modules is more a
conceptual change to facilitate better documentation and increased
adoption.
This commit is contained in:
Justin Bertram 2023-12-25 19:46:22 -06:00 committed by clebertsuconic
parent a3cde08681
commit 576622571a
110 changed files with 1016 additions and 943 deletions

View File

@ -185,12 +185,12 @@
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-quorum-api</artifactId>
<artifactId>artemis-lockmanager-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-quorum-ri</artifactId>
<artifactId>artemis-lockmanager-ri</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>

View File

@ -60,7 +60,7 @@
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-quorum-api</artifactId>
<artifactId>artemis-lockmanager-api</artifactId>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>

View File

@ -24,14 +24,14 @@ import org.apache.activemq.artemis.cli.commands.ActionContext;
import org.apache.activemq.artemis.cli.commands.tools.LockAbstract;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.HAPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedPrimitiveManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationPrimaryPolicyConfiguration;
import org.apache.activemq.artemis.core.server.NodeManager;
import org.apache.activemq.artemis.core.server.impl.FileLockNodeManager;
import org.apache.activemq.artemis.quorum.DistributedLock;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.quorum.MutableLong;
import org.apache.activemq.artemis.lockmanager.DistributedLock;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
import org.apache.activemq.artemis.lockmanager.MutableLong;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
@ -84,7 +84,7 @@ public class ActivationSequenceList extends LockAbstract {
throw new IllegalArgumentException("--node-id must be used just with --remote");
}
final HAPolicyConfiguration policyConfig = config.getHAPolicyConfiguration();
final DistributedPrimitiveManagerConfiguration managerConfiguration;
final DistributedLockManagerConfiguration managerConfiguration;
String coordinationId = null;
if (policyConfig instanceof ReplicationBackupPolicyConfiguration) {
ReplicationBackupPolicyConfiguration backupPolicyConfig = (ReplicationBackupPolicyConfiguration) policyConfig;
@ -135,7 +135,7 @@ public class ActivationSequenceList extends LockAbstract {
}
Long coordinatedSequence = null;
if (!local) {
try (DistributedPrimitiveManager manager = DistributedPrimitiveManager.newInstanceOf(
try (DistributedLockManager manager = DistributedLockManager.newInstanceOf(
managerConfiguration.getClassName(), managerConfiguration.getProperties())) {
if (!manager.start(MANAGER_START_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
throw new IllegalStateException("distributed manager isn't started in " + MANAGER_START_TIMEOUT_SECONDS + " seconds");

View File

@ -24,14 +24,14 @@ import org.apache.activemq.artemis.cli.commands.ActionContext;
import org.apache.activemq.artemis.cli.commands.tools.LockAbstract;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.HAPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedPrimitiveManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationPrimaryPolicyConfiguration;
import org.apache.activemq.artemis.core.server.NodeManager;
import org.apache.activemq.artemis.core.server.impl.FileLockNodeManager;
import org.apache.activemq.artemis.quorum.DistributedLock;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.quorum.MutableLong;
import org.apache.activemq.artemis.lockmanager.DistributedLock;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
import org.apache.activemq.artemis.lockmanager.MutableLong;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
@ -78,7 +78,7 @@ public class ActivationSequenceSet extends LockAbstract {
throw new IllegalArgumentException("--to must be >= 0");
}
final HAPolicyConfiguration policyConfig = config.getHAPolicyConfiguration();
final DistributedPrimitiveManagerConfiguration managerConfiguration;
final DistributedLockManagerConfiguration managerConfiguration;
String coordinationId = nodeId;
if (policyConfig instanceof ReplicationBackupPolicyConfiguration) {
ReplicationBackupPolicyConfiguration backupPolicyConfig = (ReplicationBackupPolicyConfiguration) policyConfig;
@ -118,7 +118,7 @@ public class ActivationSequenceSet extends LockAbstract {
}
}
if (!local) {
try (DistributedPrimitiveManager manager = DistributedPrimitiveManager.newInstanceOf(
try (DistributedLockManager manager = DistributedLockManager.newInstanceOf(
managerConfiguration.getClassName(), managerConfiguration.getProperties())) {
if (!manager.start(MANAGER_START_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
throw new IllegalStateException("distributed manager isn't started in " + MANAGER_START_TIMEOUT_SECONDS + " seconds");

View File

@ -659,7 +659,7 @@ public final class ActiveMQDefaultConfiguration {
public static final String DEFAULT_TEMPORARY_QUEUE_NAMESPACE = "";
private static final String DEFAULT_DISTRIBUTED_PRIMITIVE_MANAGER_CLASS_NAME = "org.apache.activemq.artemis.quorum.zookeeper.CuratorDistributedPrimitiveManager";
private static final String DEFAULT_DISTRIBUTED_PRIMITIVE_MANAGER_CLASS_NAME = "org.apache.activemq.artemis.lockmanager.zookeeper.CuratorDistributedLockManager";
// Number of concurrent workers for a core bridge
public static int DEFAULT_BRIDGE_CONCURRENCY = 1;
@ -1832,7 +1832,7 @@ public final class ActiveMQDefaultConfiguration {
return DEFAULT_TEMPORARY_QUEUE_NAMESPACE;
}
public static String getDefaultDistributedPrimitiveManagerClassName() {
public static String getDefaultDistributedLockManagerClassName() {
return DEFAULT_DISTRIBUTED_PRIMITIVE_MANAGER_CLASS_NAME;
}

View File

@ -130,11 +130,11 @@
<!-- quorum -->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-quorum-api</artifactId>
<artifactId>artemis-lockmanager-api</artifactId>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-quorum-ri</artifactId>
<artifactId>artemis-lockmanager-ri</artifactId>
</dependency>
<!--TODO: no other modules seem to use this, is it equivalent to something else they do use ? -->

View File

@ -80,7 +80,7 @@
<!--bundle dependency="true">mvn:io.micrometer/micrometer-core/${version.micrometer}</bundle-->
<bundle>mvn:org.apache.activemq/activemq-artemis-native/${activemq-artemis-native-version}</bundle>
<bundle>mvn:org.apache.activemq/artemis-quorum-api/${pom.version}</bundle>
<bundle>mvn:org.apache.activemq/artemis-lockmanager-api/${pom.version}</bundle>
<bundle>mvn:org.apache.activemq/artemis-server-osgi/${pom.version}</bundle>
</feature>

View File

@ -19,13 +19,15 @@
<parent>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-pom</artifactId>
<artifactId>artemis-lockmanager</artifactId>
<version>2.33.0-SNAPSHOT</version>
<relativePath>../artemis-pom/pom.xml</relativePath>
</parent>
<artifactId>artemis-quorum-api</artifactId>
<artifactId>artemis-lockmanager-api</artifactId>
<packaging>bundle</packaging>
<name>ActiveMQ Artemis Quorum API</name>
<name>ActiveMQ Artemis Lock Manager API</name>
<properties>
<activemq.basedir>${project.basedir}/../..</activemq.basedir>
</properties>
</project>

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.quorum;
package org.apache.activemq.artemis.lockmanager;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

View File

@ -14,17 +14,17 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.quorum;
package org.apache.activemq.artemis.lockmanager;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
public interface DistributedPrimitiveManager extends AutoCloseable {
public interface DistributedLockManager extends AutoCloseable {
static DistributedPrimitiveManager newInstanceOf(String className, Map<String, String> properties) throws Exception {
return (DistributedPrimitiveManager) Class.forName(className).getDeclaredConstructor(Map.class).newInstance(properties);
static DistributedLockManager newInstanceOf(String className, Map<String, String> properties) throws Exception {
return (DistributedLockManager) Class.forName(className).getDeclaredConstructor(Map.class).newInstance(properties);
}
@FunctionalInterface

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.quorum;
package org.apache.activemq.artemis.lockmanager;
public interface MutableLong extends AutoCloseable {

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.quorum;
package org.apache.activemq.artemis.lockmanager;
public final class UnavailableStateException extends Exception {

View File

@ -19,14 +19,17 @@
<parent>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-pom</artifactId>
<artifactId>artemis-lockmanager</artifactId>
<version>2.33.0-SNAPSHOT</version>
<relativePath>../artemis-pom/pom.xml</relativePath>
</parent>
<artifactId>artemis-quorum-ri</artifactId>
<artifactId>artemis-lockmanager-ri</artifactId>
<packaging>jar</packaging>
<name>ActiveMQ Artemis Quorum RI</name>
<name>ActiveMQ Artemis Lock Manager RI</name>
<properties>
<activemq.basedir>${project.basedir}/../..</activemq.basedir>
</properties>
<dependencies>
<dependency>
@ -86,7 +89,7 @@
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-quorum-api</artifactId>
<artifactId>artemis-lockmanager-api</artifactId>
</dependency>
<!-- logging -->
<dependency>

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.quorum.file;
package org.apache.activemq.artemis.lockmanager.file;
import java.io.File;
import java.io.IOException;
@ -26,27 +26,27 @@ import java.util.Objects;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import org.apache.activemq.artemis.quorum.DistributedLock;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.quorum.MutableLong;
import org.apache.activemq.artemis.quorum.UnavailableStateException;
import org.apache.activemq.artemis.lockmanager.DistributedLock;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
import org.apache.activemq.artemis.lockmanager.MutableLong;
import org.apache.activemq.artemis.lockmanager.UnavailableStateException;
/**
* This is an implementation suitable to be used just on unit tests and it won't attempt
* to manage nor purge existing stale locks files. It's part of the tests life-cycle to properly
* set-up and tear-down the environment.
*/
public class FileBasedPrimitiveManager implements DistributedPrimitiveManager {
public class FileBasedLockManager implements DistributedLockManager {
private final File locksFolder;
private final Map<String, FileDistributedLock> locks;
private boolean started;
public FileBasedPrimitiveManager(Map<String, String> args) {
public FileBasedLockManager(Map<String, String> args) {
this(new File(args.get("locks-folder")));
}
public FileBasedPrimitiveManager(File locksFolder) {
public FileBasedLockManager(File locksFolder) {
Objects.requireNonNull(locksFolder);
if (!locksFolder.exists()) {
throw new IllegalStateException(locksFolder + " is supposed to already exists");

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.quorum.file;
package org.apache.activemq.artemis.lockmanager.file;
import java.io.File;
import java.io.IOException;
@ -24,7 +24,7 @@ import java.nio.channels.OverlappingFileLockException;
import java.nio.file.StandardOpenOption;
import java.util.function.Consumer;
import org.apache.activemq.artemis.quorum.DistributedLock;
import org.apache.activemq.artemis.lockmanager.DistributedLock;
final class FileDistributedLock implements DistributedLock {

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.quorum.zookeeper;
package org.apache.activemq.artemis.lockmanager.zookeeper;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
@ -22,9 +22,9 @@ import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import org.apache.activemq.artemis.quorum.DistributedLock;
import org.apache.activemq.artemis.quorum.UnavailableStateException;
import org.apache.activemq.artemis.quorum.zookeeper.CuratorDistributedPrimitiveManager.PrimitiveId;
import org.apache.activemq.artemis.lockmanager.DistributedLock;
import org.apache.activemq.artemis.lockmanager.UnavailableStateException;
import org.apache.activemq.artemis.lockmanager.zookeeper.CuratorDistributedLockManager.PrimitiveId;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2;
import org.apache.curator.framework.recipes.locks.Lease;
@ -35,7 +35,7 @@ final class CuratorDistributedLock extends CuratorDistributedPrimitive implement
private Lease lease;
private byte[] leaseVersion;
CuratorDistributedLock(PrimitiveId id, CuratorDistributedPrimitiveManager manager, InterProcessSemaphoreV2 ipcSem) {
CuratorDistributedLock(PrimitiveId id, CuratorDistributedLockManager manager, InterProcessSemaphoreV2 ipcSem) {
super(id, manager);
this.ipcSem = ipcSem;
this.listeners = new CopyOnWriteArrayList<>();

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.quorum.zookeeper;
package org.apache.activemq.artemis.lockmanager.zookeeper;
import java.util.HashMap;
import java.util.Map;
@ -27,9 +27,9 @@ import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.activemq.artemis.quorum.DistributedLock;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.quorum.MutableLong;
import org.apache.activemq.artemis.lockmanager.DistributedLock;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
import org.apache.activemq.artemis.lockmanager.MutableLong;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
@ -43,7 +43,7 @@ import org.apache.curator.utils.DebugUtils;
import static java.util.Objects.requireNonNull;
import static java.util.stream.Collectors.joining;
public class CuratorDistributedPrimitiveManager implements DistributedPrimitiveManager, ConnectionStateListener {
public class CuratorDistributedLockManager implements DistributedLockManager, ConnectionStateListener {
enum PrimitiveType {
lock, mutableLong;
@ -154,11 +154,11 @@ public class CuratorDistributedPrimitiveManager implements DistributedPrimitiveM
}
}
public CuratorDistributedPrimitiveManager(Map<String, String> config) {
public CuratorDistributedLockManager(Map<String, String> config) {
this(validateParameters(config), true);
}
private CuratorDistributedPrimitiveManager(Map<String, String> config, boolean ignore) {
private CuratorDistributedLockManager(Map<String, String> config, boolean ignore) {
this(config.get(CONNECT_STRING_PARAM),
config.get(NAMESPACE_PARAM),
Integer.parseInt(config.getOrDefault(SESSION_MS_PARAM, DEFAULT_SESSION_TIMEOUT_MS)),
@ -168,7 +168,7 @@ public class CuratorDistributedPrimitiveManager implements DistributedPrimitiveM
Integer.parseInt(config.getOrDefault(RETRIES_MS_PARAM, DEFAULT_RETRIES_MS)));
}
private CuratorDistributedPrimitiveManager(String connectString,
private CuratorDistributedLockManager(String connectString,
String namespace,
int sessionMs,
int sessionPercent,

View File

@ -14,23 +14,23 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.quorum.zookeeper;
package org.apache.activemq.artemis.lockmanager.zookeeper;
import org.apache.activemq.artemis.quorum.UnavailableStateException;
import org.apache.activemq.artemis.quorum.zookeeper.CuratorDistributedPrimitiveManager.PrimitiveId;
import org.apache.activemq.artemis.lockmanager.UnavailableStateException;
import org.apache.activemq.artemis.lockmanager.zookeeper.CuratorDistributedLockManager.PrimitiveId;
import static org.apache.activemq.artemis.quorum.zookeeper.CuratorDistributedPrimitiveManager.PrimitiveType.validatePrimitiveInstance;
import static org.apache.activemq.artemis.lockmanager.zookeeper.CuratorDistributedLockManager.PrimitiveType.validatePrimitiveInstance;
public abstract class CuratorDistributedPrimitive implements AutoCloseable {
// this is used to prevent deadlocks on close
private final CuratorDistributedPrimitiveManager manager;
private final CuratorDistributedLockManager manager;
private final PrimitiveId id;
private boolean unavailable;
private boolean closed;
protected CuratorDistributedPrimitive(PrimitiveId id, CuratorDistributedPrimitiveManager manager) {
protected CuratorDistributedPrimitive(PrimitiveId id, CuratorDistributedLockManager manager) {
this.id = id;
this.manager = manager;
this.closed = false;

View File

@ -14,11 +14,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.quorum.zookeeper;
package org.apache.activemq.artemis.lockmanager.zookeeper;
import org.apache.activemq.artemis.quorum.MutableLong;
import org.apache.activemq.artemis.quorum.UnavailableStateException;
import org.apache.activemq.artemis.quorum.zookeeper.CuratorDistributedPrimitiveManager.PrimitiveId;
import org.apache.activemq.artemis.lockmanager.MutableLong;
import org.apache.activemq.artemis.lockmanager.UnavailableStateException;
import org.apache.activemq.artemis.lockmanager.zookeeper.CuratorDistributedLockManager.PrimitiveId;
import org.apache.curator.framework.recipes.atomic.AtomicValue;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
@ -26,7 +26,7 @@ final class CuratorMutableLong extends CuratorDistributedPrimitive implements Mu
private final DistributedAtomicLong atomicLong;
CuratorMutableLong(PrimitiveId id, CuratorDistributedPrimitiveManager manager, DistributedAtomicLong atomicLong) {
CuratorMutableLong(PrimitiveId id, CuratorDistributedLockManager manager, DistributedAtomicLong atomicLong) {
super(id, manager);
this.atomicLong = atomicLong;
}

View File

@ -0,0 +1,31 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF 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.apache.activemq.artemis.quorum.zookeeper;
import java.util.Map;
import org.apache.activemq.artemis.lockmanager.zookeeper.CuratorDistributedLockManager;
/**
* This is for backwards compatibility
*/
@Deprecated(forRemoval = true)
public class CuratorDistributedPrimitiveManager extends CuratorDistributedLockManager {
public CuratorDistributedPrimitiveManager(Map<String, String> config) {
super(config);
}
}

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.quorum;
package org.apache.activemq.artemis.lockmanager;
import java.util.ArrayList;
import java.util.HashMap;
@ -54,17 +54,17 @@ public abstract class DistributedLockTest {
});
}
protected DistributedPrimitiveManager createManagedDistributeManager() {
protected DistributedLockManager createManagedDistributeManager() {
return createManagedDistributeManager(stringStringMap -> {
});
}
protected DistributedPrimitiveManager createManagedDistributeManager(Consumer<? super Map<String, String>> defaultConfiguration) {
protected DistributedLockManager createManagedDistributeManager(Consumer<? super Map<String, String>> defaultConfiguration) {
try {
final HashMap<String, String> config = new HashMap<>();
configureManager(config);
defaultConfiguration.accept(config);
final DistributedPrimitiveManager manager = DistributedPrimitiveManager.newInstanceOf(managerClassName(), config);
final DistributedLockManager manager = DistributedLockManager.newInstanceOf(managerClassName(), config);
closeables.add(manager);
return manager;
} catch (Exception e) {
@ -74,14 +74,14 @@ public abstract class DistributedLockTest {
@Test
public void managerReturnsSameLockIfNotClosed() throws ExecutionException, InterruptedException, TimeoutException {
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
Assert.assertSame(manager.getDistributedLock("a"), manager.getDistributedLock("a"));
}
@Test
public void managerReturnsDifferentLocksIfClosed() throws ExecutionException, InterruptedException, TimeoutException {
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
DistributedLock closedLock = manager.getDistributedLock("a");
closedLock.close();
@ -90,7 +90,7 @@ public abstract class DistributedLockTest {
@Test
public void managerReturnsDifferentLocksOnRestart() throws ExecutionException, InterruptedException, TimeoutException {
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
DistributedLock closedLock = manager.getDistributedLock("a");
manager.stop();
@ -100,20 +100,20 @@ public abstract class DistributedLockTest {
@Test(expected = IllegalStateException.class)
public void managerCannotGetLockIfNotStarted() throws ExecutionException, InterruptedException, TimeoutException {
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.getDistributedLock("a");
}
@Test(expected = NullPointerException.class)
public void managerCannotGetLockWithNullLockId() throws ExecutionException, InterruptedException, TimeoutException {
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
manager.getDistributedLock(null);
}
@Test
public void closingLockUnlockIt() throws ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
DistributedLock closedLock = manager.getDistributedLock("a");
Assert.assertTrue(closedLock.tryLock());
@ -123,7 +123,7 @@ public abstract class DistributedLockTest {
@Test
public void managerStopUnlockLocks() throws ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
Assert.assertTrue(manager.getDistributedLock("a").tryLock());
Assert.assertTrue(manager.getDistributedLock("b").tryLock());
@ -135,7 +135,7 @@ public abstract class DistributedLockTest {
@Test
public void acquireAndReleaseLock() throws ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
DistributedLock lock = manager.getDistributedLock("a");
Assert.assertFalse(lock.isHeldByCaller());
@ -147,7 +147,7 @@ public abstract class DistributedLockTest {
@Test(expected = IllegalStateException.class)
public void cannotAcquireSameLockTwice() throws ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
DistributedLock lock = manager.getDistributedLock("a");
Assert.assertTrue(lock.tryLock());
@ -156,8 +156,8 @@ public abstract class DistributedLockTest {
@Test
public void heldLockIsVisibleByDifferentManagers() throws ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
DistributedPrimitiveManager ownerManager = createManagedDistributeManager();
DistributedPrimitiveManager observerManager = createManagedDistributeManager();
DistributedLockManager ownerManager = createManagedDistributeManager();
DistributedLockManager observerManager = createManagedDistributeManager();
ownerManager.start();
observerManager.start();
Assert.assertTrue(ownerManager.getDistributedLock("a").tryLock());
@ -167,8 +167,8 @@ public abstract class DistributedLockTest {
@Test
public void unlockedLockIsVisibleByDifferentManagers() throws ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
DistributedPrimitiveManager ownerManager = createManagedDistributeManager();
DistributedPrimitiveManager observerManager = createManagedDistributeManager();
DistributedLockManager ownerManager = createManagedDistributeManager();
DistributedLockManager observerManager = createManagedDistributeManager();
ownerManager.start();
observerManager.start();
Assert.assertTrue(ownerManager.getDistributedLock("a").tryLock());
@ -180,8 +180,8 @@ public abstract class DistributedLockTest {
@Test
public void cannotAcquireSameLockFromDifferentManagers() throws ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
DistributedPrimitiveManager ownerManager = createManagedDistributeManager();
DistributedPrimitiveManager notOwnerManager = createManagedDistributeManager();
DistributedLockManager ownerManager = createManagedDistributeManager();
DistributedLockManager notOwnerManager = createManagedDistributeManager();
ownerManager.start();
notOwnerManager.start();
Assert.assertTrue(ownerManager.getDistributedLock("a").tryLock());
@ -190,8 +190,8 @@ public abstract class DistributedLockTest {
@Test
public void cannotUnlockFromNotOwnerManager() throws ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
DistributedPrimitiveManager ownerManager = createManagedDistributeManager();
DistributedPrimitiveManager notOwnerManager = createManagedDistributeManager();
DistributedLockManager ownerManager = createManagedDistributeManager();
DistributedLockManager notOwnerManager = createManagedDistributeManager();
ownerManager.start();
notOwnerManager.start();
Assert.assertTrue(ownerManager.getDistributedLock("a").tryLock());
@ -202,7 +202,7 @@ public abstract class DistributedLockTest {
@Test
public void timedTryLockSucceedWithShortTimeout() throws ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
DistributedLock backgroundLock = manager.getDistributedLock("a");
Assert.assertTrue(backgroundLock.tryLock(1, TimeUnit.NANOSECONDS));
@ -210,9 +210,9 @@ public abstract class DistributedLockTest {
@Test
public void timedTryLockFailAfterTimeout() throws ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
DistributedPrimitiveManager otherManager = createManagedDistributeManager();
DistributedLockManager otherManager = createManagedDistributeManager();
otherManager.start();
Assert.assertTrue(otherManager.getDistributedLock("a").tryLock());
final long start = System.nanoTime();
@ -224,9 +224,9 @@ public abstract class DistributedLockTest {
@Test
public void timedTryLockSuccess() throws ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
DistributedPrimitiveManager otherManager = createManagedDistributeManager();
DistributedLockManager otherManager = createManagedDistributeManager();
otherManager.start();
Assert.assertTrue(otherManager.getDistributedLock("a").tryLock());
DistributedLock backgroundLock = manager.getDistributedLock("a");
@ -252,9 +252,9 @@ public abstract class DistributedLockTest {
@Test
public void interruptStopTimedTryLock() throws ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
DistributedPrimitiveManager otherManager = createManagedDistributeManager();
DistributedLockManager otherManager = createManagedDistributeManager();
otherManager.start();
Assert.assertTrue(otherManager.getDistributedLock("a").tryLock());
DistributedLock backgroundLock = manager.getDistributedLock("a");
@ -281,7 +281,7 @@ public abstract class DistributedLockTest {
@Test
public void lockAndMutableLongWithSameIdCanExistsTogether() throws ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
final String id = "a";
Assert.assertTrue(manager.getDistributedLock(id).tryLock());

View File

@ -14,15 +14,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.quorum.file;
package org.apache.activemq.artemis.lockmanager.file;
import java.io.File;
import java.lang.reflect.InvocationTargetException;
import java.util.Collections;
import java.util.Map;
import org.apache.activemq.artemis.quorum.DistributedLockTest;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.lockmanager.DistributedLockTest;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@ -49,22 +49,22 @@ public class FileDistributedLockTest extends DistributedLockTest {
@Override
protected String managerClassName() {
return FileBasedPrimitiveManager.class.getName();
return FileBasedLockManager.class.getName();
}
@Test
public void reflectiveManagerCreation() throws Exception {
DistributedPrimitiveManager.newInstanceOf(managerClassName(), Collections.singletonMap("locks-folder", locksFolder.toString()));
DistributedLockManager.newInstanceOf(managerClassName(), Collections.singletonMap("locks-folder", locksFolder.toString()));
}
@Test(expected = InvocationTargetException.class)
public void reflectiveManagerCreationFailWithoutLocksFolder() throws Exception {
DistributedPrimitiveManager.newInstanceOf(managerClassName(), Collections.emptyMap());
DistributedLockManager.newInstanceOf(managerClassName(), Collections.emptyMap());
}
@Test(expected = InvocationTargetException.class)
public void reflectiveManagerCreationFailIfLocksFolderIsNotFolder() throws Exception {
DistributedPrimitiveManager.newInstanceOf(managerClassName(), Collections.singletonMap("locks-folder", tmpFolder.newFile().toString()));
DistributedLockManager.newInstanceOf(managerClassName(), Collections.singletonMap("locks-folder", tmpFolder.newFile().toString()));
}
}

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.quorum.zookeeper;
package org.apache.activemq.artemis.lockmanager.zookeeper;
import java.util.ArrayList;
import java.util.HashMap;
@ -23,7 +23,7 @@ import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.test.InstanceSpec;
import org.apache.curator.test.TestingCluster;
@ -38,7 +38,7 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
public class CuratorDistributedPrimitiveManagerTest {
public class CuratorDistributedLockManagerTest {
private final ArrayList<AutoCloseable> autoCloseables = new ArrayList<>();
@ -88,12 +88,12 @@ public class CuratorDistributedPrimitiveManagerTest {
config.put("retries-ms", Integer.toString(RETRIES_MS));
}
protected DistributedPrimitiveManager createManagedDistributeManager(Consumer<? super Map<String, String>> defaultConfiguration) {
protected DistributedLockManager createManagedDistributeManager(Consumer<? super Map<String, String>> defaultConfiguration) {
try {
final HashMap<String, String> config = new HashMap<>();
configureManager(config);
defaultConfiguration.accept(config);
final DistributedPrimitiveManager manager = DistributedPrimitiveManager.newInstanceOf(managerClassName(), config);
final DistributedLockManager manager = DistributedLockManager.newInstanceOf(managerClassName(), config);
autoCloseables.add(manager);
return manager;
} catch (Exception e) {
@ -102,19 +102,19 @@ public class CuratorDistributedPrimitiveManagerTest {
}
protected String managerClassName() {
return CuratorDistributedPrimitiveManager.class.getName();
return CuratorDistributedLockManager.class.getName();
}
@Test
public void verifyLayoutInZK() throws Exception {
final DistributedPrimitiveManager manager = createManagedDistributeManager(config -> config.put("namespace", "activemq-artemis"));
final DistributedLockManager manager = createManagedDistributeManager(config -> config.put("namespace", "activemq-artemis"));
manager.start();
Assert.assertTrue(manager.getDistributedLock("journal-identity-000-111").tryLock());
Assert.assertTrue(manager.getMutableLong("journal-identity-000-111").compareAndSet(0, 1));
CuratorFramework curatorFramework = ((CuratorDistributedPrimitiveManager)manager).getCurator();
CuratorFramework curatorFramework = ((CuratorDistributedLockManager)manager).getCurator();
List<String> entries = new LinkedList<>();
dumpZK(curatorFramework.getZookeeperClient().getZooKeeper(), "/", entries);

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.quorum.zookeeper;
package org.apache.activemq.artemis.lockmanager.zookeeper;
import java.io.IOException;
import java.util.Arrays;
@ -28,14 +28,14 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.activemq.artemis.quorum.DistributedLock;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.quorum.UnavailableStateException;
import org.apache.activemq.artemis.lockmanager.DistributedLock;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
import org.apache.activemq.artemis.lockmanager.UnavailableStateException;
import org.apache.activemq.artemis.utils.Wait;
import org.apache.curator.test.InstanceSpec;
import org.apache.curator.test.TestingCluster;
import org.apache.activemq.artemis.quorum.DistributedLockTest;
import org.apache.activemq.artemis.lockmanager.DistributedLockTest;
import org.apache.curator.test.TestingZooKeeperServer;
import org.junit.Assert;
import org.junit.Assume;
@ -102,19 +102,19 @@ public class CuratorDistributedLockTest extends DistributedLockTest {
@Override
protected String managerClassName() {
return CuratorDistributedPrimitiveManager.class.getName();
return CuratorDistributedLockManager.class.getName();
}
@Test(expected = RuntimeException.class)
public void cannotCreateManagerWithNotValidParameterNames() {
final DistributedPrimitiveManager manager = createManagedDistributeManager(config -> config.put("_", "_"));
final DistributedLockManager manager = createManagedDistributeManager(config -> config.put("_", "_"));
}
@Test
public void canAcquireLocksFromDifferentNamespace() throws ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
final DistributedPrimitiveManager manager1 = createManagedDistributeManager(config -> config.put("namespace", "1"));
final DistributedLockManager manager1 = createManagedDistributeManager(config -> config.put("namespace", "1"));
manager1.start();
final DistributedPrimitiveManager manager2 = createManagedDistributeManager(config -> config.put("namespace", "2"));
final DistributedLockManager manager2 = createManagedDistributeManager(config -> config.put("namespace", "2"));
manager2.start();
Assert.assertTrue(manager1.getDistributedLock("a").tryLock());
Assert.assertTrue(manager2.getDistributedLock("a").tryLock());
@ -122,14 +122,14 @@ public class CuratorDistributedLockTest extends DistributedLockTest {
@Test
public void cannotStartManagerWithDisconnectedServer() throws IOException, ExecutionException, InterruptedException {
final DistributedPrimitiveManager manager = createManagedDistributeManager();
final DistributedLockManager manager = createManagedDistributeManager();
testingServer.close();
Assert.assertFalse(manager.start(1, TimeUnit.SECONDS));
}
@Test(expected = UnavailableStateException.class)
public void cannotAcquireLockWithDisconnectedServer() throws IOException, ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
final DistributedPrimitiveManager manager = createManagedDistributeManager();
final DistributedLockManager manager = createManagedDistributeManager();
manager.start();
final DistributedLock lock = manager.getDistributedLock("a");
final CountDownLatch notAvailable = new CountDownLatch(1);
@ -142,7 +142,7 @@ public class CuratorDistributedLockTest extends DistributedLockTest {
@Test(expected = UnavailableStateException.class)
public void cannotTryLockWithDisconnectedServer() throws IOException, ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
final DistributedPrimitiveManager manager = createManagedDistributeManager();
final DistributedLockManager manager = createManagedDistributeManager();
manager.start();
final DistributedLock lock = manager.getDistributedLock("a");
testingServer.close();
@ -151,7 +151,7 @@ public class CuratorDistributedLockTest extends DistributedLockTest {
@Test(expected = UnavailableStateException.class)
public void cannotCheckLockStatusWithDisconnectedServer() throws IOException, ExecutionException, InterruptedException, TimeoutException, UnavailableStateException {
final DistributedPrimitiveManager manager = createManagedDistributeManager();
final DistributedLockManager manager = createManagedDistributeManager();
manager.start();
final DistributedLock lock = manager.getDistributedLock("a");
Assert.assertFalse(lock.isHeldByCaller());
@ -162,7 +162,7 @@ public class CuratorDistributedLockTest extends DistributedLockTest {
@Test(expected = UnavailableStateException.class)
public void looseLockAfterServerStop() throws ExecutionException, InterruptedException, TimeoutException, UnavailableStateException, IOException {
final DistributedPrimitiveManager manager = createManagedDistributeManager();
final DistributedLockManager manager = createManagedDistributeManager();
manager.start();
final DistributedLock lock = manager.getDistributedLock("a");
Assert.assertTrue(lock.tryLock());
@ -179,7 +179,7 @@ public class CuratorDistributedLockTest extends DistributedLockTest {
@Test
public void canAcquireLockOnMajorityRestart() throws Exception {
Assume.assumeTrue(zkNodes + " <= 1", zkNodes > 1);
final DistributedPrimitiveManager manager = createManagedDistributeManager();
final DistributedLockManager manager = createManagedDistributeManager();
manager.start();
final DistributedLock lock = manager.getDistributedLock("a");
Assert.assertTrue(lock.tryLock());
@ -192,7 +192,7 @@ public class CuratorDistributedLockTest extends DistributedLockTest {
notAvailable.await();
manager.stop();
restartMajorityNodes(true);
final DistributedPrimitiveManager otherManager = createManagedDistributeManager();
final DistributedLockManager otherManager = createManagedDistributeManager();
otherManager.start();
// await more then the expected value, that depends by how curator session expiration is configured
TimeUnit.MILLISECONDS.sleep(SESSION_MS + SERVER_TICK_MS);
@ -202,7 +202,7 @@ public class CuratorDistributedLockTest extends DistributedLockTest {
@Test
public void cannotStartManagerWithoutQuorum() throws Exception {
Assume.assumeTrue(zkNodes + " <= 1", zkNodes > 1);
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
stopMajority(true);
Assert.assertFalse(manager.start(2, TimeUnit.SECONDS));
Assert.assertFalse(manager.isStarted());
@ -211,7 +211,7 @@ public class CuratorDistributedLockTest extends DistributedLockTest {
@Test(expected = UnavailableStateException.class)
public void cannotAcquireLockWithoutQuorum() throws Exception {
Assume.assumeTrue(zkNodes + " <= 1", zkNodes > 1);
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
stopMajority(true);
DistributedLock lock = manager.getDistributedLock("a");
@ -221,7 +221,7 @@ public class CuratorDistributedLockTest extends DistributedLockTest {
@Test
public void cannotCheckLockWithoutQuorum() throws Exception {
Assume.assumeTrue(zkNodes + " <= 1", zkNodes > 1);
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
stopMajority(true);
DistributedLock lock = manager.getDistributedLock("a");
@ -237,7 +237,7 @@ public class CuratorDistributedLockTest extends DistributedLockTest {
@Test
public void canGetLockWithoutQuorum() throws Exception {
Assume.assumeTrue(zkNodes + " <= 1", zkNodes > 1);
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
stopMajority(true);
DistributedLock lock = manager.getDistributedLock("a");
@ -247,7 +247,7 @@ public class CuratorDistributedLockTest extends DistributedLockTest {
@Test
public void notifiedAsUnavailableWhileLoosingQuorum() throws Exception {
Assume.assumeTrue(zkNodes + " <= 1", zkNodes > 1);
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
DistributedLock lock = manager.getDistributedLock("a");
CountDownLatch unavailable = new CountDownLatch(1);
@ -259,7 +259,7 @@ public class CuratorDistributedLockTest extends DistributedLockTest {
@Test
public void beNotifiedOnce() throws Exception {
Assume.assumeTrue(zkNodes + " <= 1", zkNodes > 1);
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
DistributedLock lock = manager.getDistributedLock("a");
final AtomicInteger unavailableManager = new AtomicInteger(0);
@ -275,14 +275,14 @@ public class CuratorDistributedLockTest extends DistributedLockTest {
@Test
public void beNotifiedOfUnavailabilityWhileBlockedOnTimedLock() throws Exception {
Assume.assumeTrue(zkNodes + " <= 1", zkNodes > 1);
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
DistributedLock lock = manager.getDistributedLock("a");
final AtomicInteger unavailableManager = new AtomicInteger(0);
final AtomicInteger unavailableLock = new AtomicInteger(0);
manager.addUnavailableManagerListener(unavailableManager::incrementAndGet);
lock.addListener(unavailableLock::incrementAndGet);
final DistributedPrimitiveManager otherManager = createManagedDistributeManager();
final DistributedLockManager otherManager = createManagedDistributeManager();
otherManager.start();
Assert.assertTrue(otherManager.getDistributedLock("a").tryLock());
final CountDownLatch startedTimedLock = new CountDownLatch(1);
@ -311,10 +311,10 @@ public class CuratorDistributedLockTest extends DistributedLockTest {
@Test
public void beNotifiedOfAlreadyUnavailableManagerAfterAddingListener() throws Exception {
Assume.assumeTrue(zkNodes + " <= 1", zkNodes > 1);
DistributedPrimitiveManager manager = createManagedDistributeManager();
DistributedLockManager manager = createManagedDistributeManager();
manager.start();
final AtomicBoolean unavailable = new AtomicBoolean(false);
DistributedPrimitiveManager.UnavailableManagerListener managerListener = () -> {
DistributedLockManager.UnavailableManagerListener managerListener = () -> {
unavailable.set(true);
};
manager.addUnavailableManagerListener(managerListener);

View File

@ -0,0 +1,37 @@
<!--
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
this work for additional information regarding copyright ownership.
The ASF 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.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-pom</artifactId>
<version>2.33.0-SNAPSHOT</version>
<relativePath>../artemis-pom/pom.xml</relativePath>
</parent>
<artifactId>artemis-lockmanager</artifactId>
<packaging>pom</packaging>
<name>ActiveMQ Artemis Lock Manager</name>
<modules>
<module>artemis-lockmanager-api</module>
<module>artemis-lockmanager-ri</module>
</modules>
</project>

View File

@ -88,7 +88,7 @@
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-quorum-api</artifactId>
<artifactId>artemis-lockmanager-api</artifactId>
</dependency>
<dependency>
<groupId>org.jctools</groupId>

View File

@ -75,18 +75,18 @@ public final class ConfigurationUtils {
PrimaryOnlyPolicyConfiguration pc = (PrimaryOnlyPolicyConfiguration) conf;
return new PrimaryOnlyPolicy(getScaleDownPolicy(pc.getScaleDownConfiguration()));
}
case REPLICATED: {
case REPLICATION_PRIMARY_QUORUM_VOTING: {
ReplicatedPolicyConfiguration pc = (ReplicatedPolicyConfiguration) conf;
return new ReplicatedPolicy(pc.isCheckForActiveServer(), pc.getGroupName(), pc.getClusterName(), pc.getMaxSavedReplicatedJournalsSize(), pc.getInitialReplicationSyncTimeout(), server.getNetworkHealthCheck(), pc.getVoteOnReplicationFailure(), pc.getQuorumSize(), pc.getVoteRetries(), pc.getVoteRetryWait(), pc.getQuorumVoteWait(), pc.getRetryReplicationWait());
}
case REPLICA: {
case REPLICATION_BACKUP_QUORUM_VOTING: {
ReplicaPolicyConfiguration pc = (ReplicaPolicyConfiguration) conf;
return new ReplicaPolicy(pc.getClusterName(), pc.getMaxSavedReplicatedJournalsSize(), pc.getGroupName(), pc.isRestartBackup(), pc.isAllowFailBack(), pc.getInitialReplicationSyncTimeout(), getScaleDownPolicy(pc.getScaleDownConfiguration()), server.getNetworkHealthCheck(), pc.getVoteOnReplicationFailure(), pc.getQuorumSize(), pc.getVoteRetries(), pc.getVoteRetryWait(), pc.getQuorumVoteWait(), pc.getRetryReplicationWait());
}
case REPLICATION_PRIMARY: {
case REPLICATION_PRIMARY_LOCK_MANAGER: {
return ReplicationPrimaryPolicy.with((ReplicationPrimaryPolicyConfiguration) conf);
}
case REPLICATION_BACKUP: {
case REPLICATION_BACKUP_LOCK_MANAGER: {
return ReplicationBackupPolicy.with((ReplicationBackupPolicyConfiguration) conf);
}
case SHARED_STORE_PRIMARY: {

View File

@ -22,13 +22,13 @@ public interface HAPolicyConfiguration extends Serializable {
enum TYPE {
PRIMARY_ONLY("Primary Only"),
REPLICATED("Replicated"),
REPLICA("Replica"),
REPLICATION_PRIMARY_QUORUM_VOTING("Replication Primary w/quorum voting"),
REPLICATION_BACKUP_QUORUM_VOTING("Replication Backup w/quorum voting"),
SHARED_STORE_PRIMARY("Shared Store Primary"),
SHARED_STORE_BACKUP("Shared Store Backup"),
COLOCATED("Colocated"),
REPLICATION_PRIMARY("Replication Primary w/pluggable quorum voting"),
REPLICATION_BACKUP("Replication Backup w/pluggable quorum voting");
REPLICATION_PRIMARY_LOCK_MANAGER("Replication Primary w/lock manager"),
REPLICATION_BACKUP_LOCK_MANAGER("Replication Backup w/lock manager");
private String name;

View File

@ -20,16 +20,16 @@ import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
public class DistributedPrimitiveManagerConfiguration implements Serializable {
public class DistributedLockManagerConfiguration implements Serializable {
private String className;
private final Map<String, String> properties;
public DistributedPrimitiveManagerConfiguration() {
public DistributedLockManagerConfiguration() {
properties = new HashMap<>();
}
public DistributedPrimitiveManagerConfiguration(String className, Map<String, String> properties) {
public DistributedLockManagerConfiguration(String className, Map<String, String> properties) {
this.className = className;
this.properties = properties;
}
@ -38,7 +38,7 @@ public class DistributedPrimitiveManagerConfiguration implements Serializable {
return className;
}
public DistributedPrimitiveManagerConfiguration setClassName(String className) {
public DistributedLockManagerConfiguration setClassName(String className) {
this.className = className;
return this;
}

View File

@ -57,7 +57,7 @@ public class ReplicaPolicyConfiguration implements HAPolicyConfiguration {
@Override
public TYPE getType() {
return TYPE.REPLICA;
return TYPE.REPLICATION_BACKUP_QUORUM_VOTING;
}
public ScaleDownConfiguration getScaleDownConfiguration() {

View File

@ -49,7 +49,7 @@ public class ReplicatedPolicyConfiguration implements HAPolicyConfiguration {
@Override
public TYPE getType() {
return TYPE.REPLICATED;
return TYPE.REPLICATION_PRIMARY_QUORUM_VOTING;
}
public boolean isCheckForActiveServer() {

View File

@ -36,7 +36,7 @@ public class ReplicationBackupPolicyConfiguration implements HAPolicyConfigurati
private long retryReplicationWait = ActiveMQDefaultConfiguration.getDefaultRetryReplicationWait();
private DistributedPrimitiveManagerConfiguration distributedManagerConfiguration = null;
private DistributedLockManagerConfiguration distributedManagerConfiguration = null;
public static final ReplicationBackupPolicyConfiguration withDefault() {
return new ReplicationBackupPolicyConfiguration();
@ -47,7 +47,7 @@ public class ReplicationBackupPolicyConfiguration implements HAPolicyConfigurati
@Override
public HAPolicyConfiguration.TYPE getType() {
return TYPE.REPLICATION_BACKUP;
return TYPE.REPLICATION_BACKUP_LOCK_MANAGER;
}
public String getClusterName() {
@ -104,12 +104,12 @@ public class ReplicationBackupPolicyConfiguration implements HAPolicyConfigurati
return this;
}
public ReplicationBackupPolicyConfiguration setDistributedManagerConfiguration(DistributedPrimitiveManagerConfiguration configuration) {
public ReplicationBackupPolicyConfiguration setDistributedManagerConfiguration(DistributedLockManagerConfiguration configuration) {
this.distributedManagerConfiguration = configuration;
return this;
}
public DistributedPrimitiveManagerConfiguration getDistributedManagerConfiguration() {
public DistributedLockManagerConfiguration getDistributedManagerConfiguration() {
return distributedManagerConfiguration;
}
}

View File

@ -29,7 +29,7 @@ public class ReplicationPrimaryPolicyConfiguration implements HAPolicyConfigurat
private Long retryReplicationWait = ActiveMQDefaultConfiguration.getDefaultRetryReplicationWait();
private DistributedPrimitiveManagerConfiguration distributedManagerConfiguration = null;
private DistributedLockManagerConfiguration distributedManagerConfiguration = null;
private String coordinationId = null;
@ -44,7 +44,7 @@ public class ReplicationPrimaryPolicyConfiguration implements HAPolicyConfigurat
@Override
public TYPE getType() {
return TYPE.REPLICATION_PRIMARY;
return TYPE.REPLICATION_PRIMARY_LOCK_MANAGER;
}
public String getGroupName() {
@ -82,12 +82,12 @@ public class ReplicationPrimaryPolicyConfiguration implements HAPolicyConfigurat
return retryReplicationWait;
}
public ReplicationPrimaryPolicyConfiguration setDistributedManagerConfiguration(DistributedPrimitiveManagerConfiguration configuration) {
public ReplicationPrimaryPolicyConfiguration setDistributedManagerConfiguration(DistributedLockManagerConfiguration configuration) {
this.distributedManagerConfiguration = configuration;
return this;
}
public DistributedPrimitiveManagerConfiguration getDistributedManagerConfiguration() {
public DistributedLockManagerConfiguration getDistributedManagerConfiguration() {
return distributedManagerConfiguration;
}

View File

@ -29,6 +29,7 @@ import java.io.ObjectOutputStream;
import java.io.PrintWriter;
import java.io.Serializable;
import java.io.StringWriter;
import java.lang.invoke.MethodHandles;
import java.lang.reflect.Array;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
@ -51,6 +52,8 @@ import java.util.Set;
import java.util.Stack;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.zip.Adler32;
import java.util.zip.Checksum;
import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
import org.apache.activemq.artemis.api.core.BroadcastGroupConfiguration;
@ -60,16 +63,6 @@ import org.apache.activemq.artemis.api.core.Pair;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.core.config.TransformerConfiguration;
import org.apache.activemq.artemis.core.config.ha.ColocatedPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.LiveOnlyPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationPrimaryPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.SharedStoreBackupPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.SharedStorePrimaryPolicyConfiguration;
import org.apache.activemq.artemis.core.config.routing.ConnectionRouterConfiguration;
import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectConfiguration;
import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPFederationBrokerPlugin;
import org.apache.activemq.artemis.core.config.BridgeConfiguration;
import org.apache.activemq.artemis.core.config.ClusterConnectionConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
@ -82,16 +75,26 @@ import org.apache.activemq.artemis.core.config.FederationConfiguration;
import org.apache.activemq.artemis.core.config.HAPolicyConfiguration;
import org.apache.activemq.artemis.core.config.MetricsConfiguration;
import org.apache.activemq.artemis.core.config.StoreConfiguration;
import org.apache.activemq.artemis.core.config.TransformerConfiguration;
import org.apache.activemq.artemis.core.config.WildcardConfiguration;
import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectConfiguration;
import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPFederationBrokerPlugin;
import org.apache.activemq.artemis.core.config.ha.ColocatedPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.LiveOnlyPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicaPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicatedPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationPrimaryPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.SharedStoreBackupPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.SharedStorePrimaryPolicyConfiguration;
import org.apache.activemq.artemis.core.config.routing.ConnectionRouterConfiguration;
import org.apache.activemq.artemis.core.config.routing.NamedPropertyConfiguration;
import org.apache.activemq.artemis.core.config.storage.DatabaseStorageConfiguration;
import org.apache.activemq.artemis.core.remoting.impl.invm.InVMConnectorFactory;
import org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnectorFactory;
import org.apache.activemq.artemis.core.security.Role;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.server.ActiveMQMessageBundle;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.server.JournalType;
import org.apache.activemq.artemis.core.server.NetworkHealthCheck;
import org.apache.activemq.artemis.core.server.SecuritySettingPlugin;
@ -130,13 +133,9 @@ import org.apache.commons.beanutils.MappedPropertyDescriptor;
import org.apache.commons.beanutils.MethodUtils;
import org.apache.commons.beanutils.PropertyUtilsBean;
import org.apache.commons.beanutils.expression.DefaultResolver;
import org.apache.commons.beanutils.expression.Resolver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.invoke.MethodHandles;
import java.util.zip.Adler32;
import java.util.zip.Checksum;
import org.apache.commons.beanutils.expression.Resolver;
public class ConfigurationImpl implements Configuration, Serializable {
@ -884,9 +883,9 @@ public class ConfigurationImpl implements Configuration, Serializable {
switch (haPolicyType) {
case PRIMARY_ONLY:
return (T) new LiveOnlyPolicyConfiguration();
case REPLICATED:
case REPLICATION_PRIMARY_QUORUM_VOTING:
return (T) new ReplicatedPolicyConfiguration();
case REPLICA:
case REPLICATION_BACKUP_QUORUM_VOTING:
return (T) new ReplicaPolicyConfiguration();
case SHARED_STORE_PRIMARY:
return (T) new SharedStorePrimaryPolicyConfiguration();
@ -894,9 +893,9 @@ public class ConfigurationImpl implements Configuration, Serializable {
return (T) new SharedStoreBackupPolicyConfiguration();
case COLOCATED:
return (T) new ColocatedPolicyConfiguration();
case REPLICATION_PRIMARY:
case REPLICATION_PRIMARY_LOCK_MANAGER:
return (T) ReplicationPrimaryPolicyConfiguration.withDefault();
case REPLICATION_BACKUP:
case REPLICATION_BACKUP_LOCK_MANAGER:
return (T) ReplicationBackupPolicyConfiguration.withDefault();
}

View File

@ -71,7 +71,7 @@ import org.apache.activemq.artemis.core.config.federation.FederationStreamConfig
import org.apache.activemq.artemis.core.config.federation.FederationTransformerConfiguration;
import org.apache.activemq.artemis.core.config.federation.FederationUpstreamConfiguration;
import org.apache.activemq.artemis.core.config.ha.ColocatedPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedPrimitiveManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.PrimaryOnlyPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicaPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicatedPolicyConfiguration;
@ -1823,7 +1823,7 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
configuration.setClusterName(getString(policyNode, "cluster-name", configuration.getClusterName(), NO_CHECK));
configuration.setInitialReplicationSyncTimeout(getLong(policyNode, "initial-replication-sync-timeout", configuration.getInitialReplicationSyncTimeout(), GT_ZERO));
configuration.setRetryReplicationWait(getLong(policyNode, "retry-replication-wait", configuration.getRetryReplicationWait(), GT_ZERO));
configuration.setDistributedManagerConfiguration(createDistributedPrimitiveManagerConfiguration(policyNode));
configuration.setDistributedManagerConfiguration(createDistributedLockManagerConfiguration(policyNode));
configuration.setCoordinationId(getString(policyNode, "coordination-id", configuration.getCoordinationId(), NOT_NULL_OR_EMPTY));
configuration.setMaxSavedReplicatedJournalsSize(getInteger(policyNode, "max-saved-replicated-journals-size", configuration.getMaxSavedReplicatedJournalsSize(), MINUS_ONE_OR_GE_ZERO));
return configuration;
@ -1837,14 +1837,14 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
configuration.setClusterName(getString(policyNode, "cluster-name", configuration.getClusterName(), NO_CHECK));
configuration.setMaxSavedReplicatedJournalsSize(getInteger(policyNode, "max-saved-replicated-journals-size", configuration.getMaxSavedReplicatedJournalsSize(), MINUS_ONE_OR_GE_ZERO));
configuration.setRetryReplicationWait(getLong(policyNode, "retry-replication-wait", configuration.getRetryReplicationWait(), GT_ZERO));
configuration.setDistributedManagerConfiguration(createDistributedPrimitiveManagerConfiguration(policyNode));
configuration.setDistributedManagerConfiguration(createDistributedLockManagerConfiguration(policyNode));
return configuration;
}
private DistributedPrimitiveManagerConfiguration createDistributedPrimitiveManagerConfiguration(Element policyNode) {
private DistributedLockManagerConfiguration createDistributedLockManagerConfiguration(Element policyNode) {
final Element managerNode = (Element) policyNode.getElementsByTagName("manager").item(0);
final String className = getString(managerNode, "class-name",
ActiveMQDefaultConfiguration.getDefaultDistributedPrimitiveManagerClassName(),
ActiveMQDefaultConfiguration.getDefaultDistributedLockManagerClassName(),
NO_CHECK);
final Map<String, String> properties;
if (parameterExists(managerNode, "properties")) {
@ -1860,7 +1860,7 @@ public final class FileConfigurationParser extends XMLConfigurationUtil {
} else {
properties = new HashMap<>(1);
}
return new DistributedPrimitiveManagerConfiguration(className, properties);
return new DistributedLockManagerConfiguration(className, properties);
}
private SharedStorePrimaryPolicyConfiguration createSharedStorePrimaryHaPolicy(Element policyNode) {

View File

@ -20,8 +20,8 @@ import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
import org.apache.activemq.artemis.core.server.cluster.qourum.QuorumVoteHandler;
import org.apache.activemq.artemis.core.server.cluster.qourum.Vote;
import org.apache.activemq.artemis.core.server.cluster.quorum.QuorumVoteHandler;
import org.apache.activemq.artemis.core.server.cluster.quorum.Vote;
public class QuorumVoteMessage extends PacketImpl {

View File

@ -20,8 +20,8 @@ import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
import org.apache.activemq.artemis.core.server.cluster.qourum.QuorumVoteHandler;
import org.apache.activemq.artemis.core.server.cluster.qourum.Vote;
import org.apache.activemq.artemis.core.server.cluster.quorum.QuorumVoteHandler;
import org.apache.activemq.artemis.core.server.cluster.quorum.Vote;
public class QuorumVoteReplyMessage extends PacketImpl {

View File

@ -80,7 +80,7 @@ import org.apache.activemq.artemis.core.server.ActiveMQMessageBundle;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.server.cluster.ClusterManager;
import org.apache.activemq.artemis.core.server.cluster.qourum.QuorumManager;
import org.apache.activemq.artemis.core.server.cluster.quorum.QuorumManager;
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
import org.apache.activemq.artemis.spi.core.remoting.Connection;
import org.apache.activemq.artemis.spi.core.remoting.ReadyListener;

View File

@ -44,7 +44,7 @@ import org.apache.activemq.artemis.core.server.routing.targets.Target;
import org.apache.activemq.artemis.core.server.cluster.Bridge;
import org.apache.activemq.artemis.core.server.cluster.impl.BridgeImpl;
import org.apache.activemq.artemis.core.server.cluster.impl.ClusterConnectionImpl;
import org.apache.activemq.artemis.core.server.cluster.qourum.ServerConnectVote;
import org.apache.activemq.artemis.core.server.cluster.quorum.ServerConnectVote;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.core.server.impl.ServerSessionImpl;
import org.apache.activemq.artemis.core.server.management.Notification;
@ -68,7 +68,7 @@ public interface ActiveMQServerLogger {
@LogMessage(id = 223001, value = "Ignored quorum vote due to quorum reached or vote casted: {}", level = LogMessage.Level.DEBUG)
void ignoredQuorumVote(ServerConnectVote vote);
@LogMessage(id = 221000, value = "{} Message Broker is starting with configuration {}", level = LogMessage.Level.INFO)
@LogMessage(id = 221000, value = "{} message broker is starting with configuration {}", level = LogMessage.Level.INFO)
void serverStarting(String type, Configuration configuration);
@LogMessage(id = 221001, value = "Apache ActiveMQ Artemis Message Broker version {} [{}, nodeID={}] {}", level = LogMessage.Level.INFO)

View File

@ -49,7 +49,7 @@ import org.apache.activemq.artemis.core.protocol.core.impl.wireformat.ScaleDownA
import org.apache.activemq.artemis.core.server.ActiveMQComponent;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.server.cluster.qourum.QuorumManager;
import org.apache.activemq.artemis.core.server.cluster.quorum.QuorumManager;
import org.apache.activemq.artemis.core.server.impl.Activation;
import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
import org.slf4j.Logger;

View File

@ -57,7 +57,7 @@ import org.apache.activemq.artemis.core.server.cluster.ha.HAManager;
import org.apache.activemq.artemis.core.server.cluster.impl.BridgeImpl;
import org.apache.activemq.artemis.core.server.cluster.impl.BroadcastGroupImpl;
import org.apache.activemq.artemis.core.server.cluster.impl.ClusterConnectionImpl;
import org.apache.activemq.artemis.core.server.cluster.qourum.QuorumManager;
import org.apache.activemq.artemis.core.server.cluster.quorum.QuorumManager;
import org.apache.activemq.artemis.core.server.impl.Activation;
import org.apache.activemq.artemis.core.server.management.ManagementService;
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;

View File

@ -20,11 +20,11 @@ import java.util.Map;
import java.util.Objects;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedPrimitiveManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration;
import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.core.server.impl.ReplicationBackupActivation;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
public class ReplicationBackupPolicy implements HAPolicy<ReplicationBackupActivation> {
@ -33,7 +33,7 @@ public class ReplicationBackupPolicy implements HAPolicy<ReplicationBackupActiva
private final String clusterName;
private final int maxSavedReplicatedJournalsSize;
private final long retryReplicationWait;
private final DistributedPrimitiveManagerConfiguration managerConfiguration;
private final DistributedLockManagerConfiguration managerConfiguration;
private final boolean tryFailback;
private ReplicationBackupPolicy(ReplicationBackupPolicyConfiguration configuration,
@ -83,7 +83,7 @@ public class ReplicationBackupPolicy implements HAPolicy<ReplicationBackupActiva
String clusterName,
String groupName,
ReplicationPrimaryPolicy primaryPolicy,
DistributedPrimitiveManagerConfiguration distributedManagerConfiguration) {
DistributedLockManagerConfiguration distributedManagerConfiguration) {
return new ReplicationBackupPolicy(ReplicationBackupPolicyConfiguration.withDefault()
.setRetryReplicationWait(retryReplicationWait)
.setMaxSavedReplicatedJournalsSize(maxSavedReplicatedJournalsSize)
@ -98,7 +98,7 @@ public class ReplicationBackupPolicy implements HAPolicy<ReplicationBackupActiva
boolean wasPrimary,
Map<String, Object> activationParams,
IOCriticalErrorListener shutdownOnCriticalIO) throws Exception {
return new ReplicationBackupActivation(server, DistributedPrimitiveManager.newInstanceOf(
return new ReplicationBackupActivation(server, DistributedLockManager.newInstanceOf(
managerConfiguration.getClassName(), managerConfiguration.getProperties()), this);
}

View File

@ -20,11 +20,11 @@ import java.util.Map;
import java.util.Objects;
import org.apache.activemq.artemis.core.config.ha.ReplicationPrimaryPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedPrimitiveManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration;
import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.core.server.impl.ReplicationPrimaryActivation;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
public class ReplicationPrimaryPolicy implements HAPolicy<ReplicationPrimaryActivation> {
@ -32,7 +32,7 @@ public class ReplicationPrimaryPolicy implements HAPolicy<ReplicationPrimaryActi
private final String clusterName;
private final String groupName;
private final long initialReplicationSyncTimeout;
private final DistributedPrimitiveManagerConfiguration distributedManagerConfiguration;
private final DistributedLockManagerConfiguration distributedManagerConfiguration;
private final boolean allowAutoFailBack;
private final String coordinationId;
@ -70,7 +70,7 @@ public class ReplicationPrimaryPolicy implements HAPolicy<ReplicationPrimaryActi
String clusterName,
ReplicationBackupPolicy replicaPolicy,
boolean allowAutoFailback,
DistributedPrimitiveManagerConfiguration distributedManagerConfiguration) {
DistributedLockManagerConfiguration distributedManagerConfiguration) {
return new ReplicationPrimaryPolicy(ReplicationPrimaryPolicyConfiguration.withDefault()
.setInitialReplicationSyncTimeout(initialReplicationSyncTimeout)
.setGroupName(groupName)
@ -97,7 +97,7 @@ public class ReplicationPrimaryPolicy implements HAPolicy<ReplicationPrimaryActi
Map<String, Object> activationParams,
IOCriticalErrorListener shutdownOnCriticalIO) throws Exception {
return new ReplicationPrimaryActivation(server,
DistributedPrimitiveManager.newInstanceOf(
DistributedLockManager.newInstanceOf(
distributedManagerConfiguration.getClassName(),
distributedManagerConfiguration.getProperties()), this);
}

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.core.server.cluster.qourum;
package org.apache.activemq.artemis.core.server.cluster.quorum;
import java.util.Map;

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.core.server.cluster.qourum;
package org.apache.activemq.artemis.core.server.cluster.quorum;
import org.apache.activemq.artemis.core.client.impl.Topology;

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.core.server.cluster.qourum;
package org.apache.activemq.artemis.core.server.cluster.quorum;
import java.util.ArrayList;
import java.util.HashMap;
@ -45,8 +45,8 @@ import org.apache.activemq.artemis.core.server.cluster.ClusterController;
import static org.apache.activemq.artemis.utils.Preconditions.checkNotNull;
/**
* A QourumManager can be used to register a {@link org.apache.activemq.artemis.core.server.cluster.qourum.Quorum} to receive notifications
* about changes to the cluster. A {@link org.apache.activemq.artemis.core.server.cluster.qourum.Quorum} can then issue a vote to the
* A QourumManager can be used to register a {@link org.apache.activemq.artemis.core.server.cluster.quorum.Quorum} to receive notifications
* about changes to the cluster. A {@link org.apache.activemq.artemis.core.server.cluster.quorum.Quorum} can then issue a vote to the
* remaining nodes in a cluster for a specific outcome
*/
public final class QuorumManager implements ClusterTopologyListener, ActiveMQComponent {
@ -56,7 +56,7 @@ public final class QuorumManager implements ClusterTopologyListener, ActiveMQCom
private final ClusterController clusterController;
/**
* all the current registered {@link org.apache.activemq.artemis.core.server.cluster.qourum.Quorum}'s
* all the current registered {@link org.apache.activemq.artemis.core.server.cluster.quorum.Quorum}'s
*/
private final Map<String, Quorum> quorums = new HashMap<>();
@ -128,7 +128,7 @@ public final class QuorumManager implements ClusterTopologyListener, ActiveMQCom
}
/**
* registers a {@link org.apache.activemq.artemis.core.server.cluster.qourum.Quorum} so that it can be notified of changes in the cluster.
* registers a {@link org.apache.activemq.artemis.core.server.cluster.quorum.Quorum} so that it can be notified of changes in the cluster.
*
* @param quorum
*/
@ -138,7 +138,7 @@ public final class QuorumManager implements ClusterTopologyListener, ActiveMQCom
}
/**
* unregisters a {@link org.apache.activemq.artemis.core.server.cluster.qourum.Quorum}.
* unregisters a {@link org.apache.activemq.artemis.core.server.cluster.quorum.Quorum}.
*
* @param quorum
*/
@ -148,7 +148,7 @@ public final class QuorumManager implements ClusterTopologyListener, ActiveMQCom
/**
* called by the {@link org.apache.activemq.artemis.core.client.impl.ServerLocatorInternal} when the topology changes. we update the
* {@code maxClusterSize} if needed and inform the {@link org.apache.activemq.artemis.core.server.cluster.qourum.Quorum}'s.
* {@code maxClusterSize} if needed and inform the {@link org.apache.activemq.artemis.core.server.cluster.quorum.Quorum}'s.
*
* @param topologyMember the topolgy changed
* @param last if the whole cluster topology is being transmitted (after adding the listener to
@ -164,7 +164,7 @@ public final class QuorumManager implements ClusterTopologyListener, ActiveMQCom
}
/**
* notify the {@link org.apache.activemq.artemis.core.server.cluster.qourum.Quorum} of a topology change.
* notify the {@link org.apache.activemq.artemis.core.server.cluster.quorum.Quorum} of a topology change.
*
* @param eventUID
* @param nodeID the id of the node leaving the cluster
@ -367,7 +367,7 @@ public final class QuorumManager implements ClusterTopologyListener, ActiveMQCom
/**
* this will connect to a node and then cast a vote. whether or not this vote is asked of the target node is dependent
* on {@link org.apache.activemq.artemis.core.server.cluster.qourum.Vote#isRequestServerVote()}
* on {@link org.apache.activemq.artemis.core.server.cluster.quorum.Vote#isRequestServerVote()}
*/
private final class VoteRunnable implements Runnable {

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.core.server.cluster.qourum;
package org.apache.activemq.artemis.core.server.cluster.quorum;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.client.impl.Topology;
@ -31,7 +31,7 @@ public abstract class QuorumVote<V extends Vote, T> {
}
/**
* called by the {@link org.apache.activemq.artemis.core.server.cluster.qourum.QuorumManager} when one of the nodes in the quorum is
* called by the {@link org.apache.activemq.artemis.core.server.cluster.quorum.QuorumManager} when one of the nodes in the quorum is
* successfully connected to. The QuorumVote can then decide whether or not a decision can be made with just that information.
*
* @return the vote to use
@ -39,7 +39,7 @@ public abstract class QuorumVote<V extends Vote, T> {
public abstract Vote connected();
/**
* called by the {@link org.apache.activemq.artemis.core.server.cluster.qourum.QuorumManager} fails to connect to a node in the quorum.
* called by the {@link org.apache.activemq.artemis.core.server.cluster.quorum.QuorumManager} fails to connect to a node in the quorum.
* The QuorumVote can then decide whether or not a decision can be made with just that information however the node
* cannot cannot be asked.
*
@ -48,7 +48,7 @@ public abstract class QuorumVote<V extends Vote, T> {
public abstract Vote notConnected();
/**
* called by the {@link org.apache.activemq.artemis.core.server.cluster.qourum.QuorumManager} when a vote can be made, either from the
* called by the {@link org.apache.activemq.artemis.core.server.cluster.quorum.QuorumManager} when a vote can be made, either from the
* cluster or decided by itself.
*
* @param vote the vote to make.
@ -63,14 +63,14 @@ public abstract class QuorumVote<V extends Vote, T> {
public abstract T getDecision();
/**
* called by the {@link org.apache.activemq.artemis.core.server.cluster.qourum.QuorumManager} when all the votes have been cast and received.
* called by the {@link org.apache.activemq.artemis.core.server.cluster.quorum.QuorumManager} when all the votes have been cast and received.
*
* @param voteTopology the topology of where the votes were sent.
*/
public abstract void allVotesCast(Topology voteTopology);
/**
* the name of this quorum vote, used for identifying the correct {@link org.apache.activemq.artemis.core.server.cluster.qourum.QuorumVoteHandler}
* the name of this quorum vote, used for identifying the correct {@link org.apache.activemq.artemis.core.server.cluster.quorum.QuorumVoteHandler}
*
* @return the name of the wuorum vote
*/

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.core.server.cluster.qourum;
package org.apache.activemq.artemis.core.server.cluster.quorum;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.SimpleString;

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.core.server.cluster.qourum;
package org.apache.activemq.artemis.core.server.cluster.quorum;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.core.server.cluster.qourum;
package org.apache.activemq.artemis.core.server.cluster.quorum;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.core.server.cluster.qourum;
package org.apache.activemq.artemis.core.server.cluster.quorum;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledExecutorService;

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.core.server.cluster.qourum;
package org.apache.activemq.artemis.core.server.cluster.quorum;
import java.util.HashMap;
import java.util.Map;

View File

@ -718,7 +718,7 @@ public class ActiveMQServerImpl implements ActiveMQServer {
throw e;
}
ActiveMQServerLogger.LOGGER.serverStarting((haPolicy.isBackup() ? "backup" : "primary"), configuration);
ActiveMQServerLogger.LOGGER.serverStarting((haPolicy.isBackup() ? "Backup" : "Primary"), configuration);
final boolean wasPrimary = !haPolicy.isBackup();
if (!haPolicy.isBackup()) {

View File

@ -39,9 +39,9 @@ import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.server.cluster.ha.ColocatedHAManager;
import org.apache.activemq.artemis.core.server.cluster.ha.ColocatedPolicy;
import org.apache.activemq.artemis.core.server.cluster.ha.HAManager;
import org.apache.activemq.artemis.core.server.cluster.qourum.QuorumVote;
import org.apache.activemq.artemis.core.server.cluster.qourum.QuorumVoteHandler;
import org.apache.activemq.artemis.core.server.cluster.qourum.Vote;
import org.apache.activemq.artemis.core.server.cluster.quorum.QuorumVote;
import org.apache.activemq.artemis.core.server.cluster.quorum.QuorumVoteHandler;
import org.apache.activemq.artemis.core.server.cluster.quorum.Vote;
import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
public class ColocatedActivation extends PrimaryActivation {

View File

@ -36,9 +36,9 @@ import org.apache.activemq.artemis.core.server.NodeManager;
import org.apache.activemq.artemis.core.server.cluster.ClusterControl;
import org.apache.activemq.artemis.core.server.cluster.ClusterController;
import org.apache.activemq.artemis.core.server.cluster.ha.ReplicationBackupPolicy;
import org.apache.activemq.artemis.quorum.DistributedLock;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.quorum.UnavailableStateException;
import org.apache.activemq.artemis.lockmanager.DistributedLock;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
import org.apache.activemq.artemis.lockmanager.UnavailableStateException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.invoke.MethodHandles;
@ -52,7 +52,7 @@ import static org.apache.activemq.artemis.core.server.impl.quorum.ActivationSequ
* This activation can be used by a primary while trying to fail-back ie {@code failback == true} or
* by a natural-born backup ie {@code failback == false}.<br>
*/
public final class ReplicationBackupActivation extends Activation implements DistributedPrimitiveManager.UnavailableManagerListener {
public final class ReplicationBackupActivation extends Activation implements DistributedLockManager.UnavailableManagerListener {
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@ -62,7 +62,7 @@ public final class ReplicationBackupActivation extends Activation implements Dis
private final String expectedNodeID;
@GuardedBy("this")
private boolean closed;
private final DistributedPrimitiveManager distributedManager;
private final DistributedLockManager distributedManager;
// Used for monitoring purposes
private volatile ReplicationObserver replicationObserver;
// Used for testing purposes
@ -73,7 +73,7 @@ public final class ReplicationBackupActivation extends Activation implements Dis
private final AtomicBoolean stopping;
public ReplicationBackupActivation(final ActiveMQServerImpl activeMQServer,
final DistributedPrimitiveManager distributedManager,
final DistributedLockManager distributedManager,
final ReplicationBackupPolicy policy) {
this.activeMQServer = activeMQServer;
if (policy.isTryFailback()) {
@ -101,7 +101,7 @@ public final class ReplicationBackupActivation extends Activation implements Dis
/**
* used for testing purposes.
*/
public DistributedPrimitiveManager getDistributedManager() {
public DistributedLockManager getDistributedManager() {
return distributedManager;
}

View File

@ -43,9 +43,9 @@ import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.server.NodeManager;
import org.apache.activemq.artemis.core.server.cluster.ClusterConnection;
import org.apache.activemq.artemis.core.server.cluster.ha.ReplicationPrimaryPolicy;
import org.apache.activemq.artemis.quorum.DistributedLock;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.quorum.UnavailableStateException;
import org.apache.activemq.artemis.lockmanager.DistributedLock;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
import org.apache.activemq.artemis.lockmanager.UnavailableStateException;
import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -77,12 +77,12 @@ public class ReplicationPrimaryActivation extends PrimaryActivation implements D
private final Object replicationLock;
private final DistributedPrimitiveManager distributedManager;
private final DistributedLockManager distributedManager;
private final AtomicBoolean stoppingServer;
public ReplicationPrimaryActivation(final ActiveMQServerImpl activeMQServer,
final DistributedPrimitiveManager distributedManager,
final DistributedLockManager distributedManager,
final ReplicationPrimaryPolicy policy) {
this.activeMQServer = activeMQServer;
this.policy = policy;
@ -94,7 +94,7 @@ public class ReplicationPrimaryActivation extends PrimaryActivation implements D
/**
* used for testing purposes.
*/
public DistributedPrimitiveManager getDistributedManager() {
public DistributedLockManager getDistributedManager() {
return distributedManager;
}

View File

@ -20,10 +20,10 @@ import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.client.impl.TopologyMemberImpl;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.server.cluster.qourum.QuorumVoteHandler;
import org.apache.activemq.artemis.core.server.cluster.qourum.QuorumVoteServerConnect;
import org.apache.activemq.artemis.core.server.cluster.qourum.ServerConnectVote;
import org.apache.activemq.artemis.core.server.cluster.qourum.Vote;
import org.apache.activemq.artemis.core.server.cluster.quorum.QuorumVoteHandler;
import org.apache.activemq.artemis.core.server.cluster.quorum.QuorumVoteServerConnect;
import org.apache.activemq.artemis.core.server.cluster.quorum.ServerConnectVote;
import org.apache.activemq.artemis.core.server.cluster.quorum.Vote;
public class ServerConnectVoteHandler implements QuorumVoteHandler {
private final ActiveMQServerImpl server;

View File

@ -45,7 +45,7 @@ import org.apache.activemq.artemis.core.server.cluster.ClusterControl;
import org.apache.activemq.artemis.core.server.cluster.ClusterController;
import org.apache.activemq.artemis.core.server.cluster.ha.ReplicaPolicy;
import org.apache.activemq.artemis.core.server.cluster.ha.ScaleDownPolicy;
import org.apache.activemq.artemis.core.server.cluster.qourum.SharedNothingBackupQuorum;
import org.apache.activemq.artemis.core.server.cluster.quorum.SharedNothingBackupQuorum;
import org.apache.activemq.artemis.core.server.group.GroupingHandler;
import org.apache.activemq.artemis.core.server.management.ManagementService;
import org.apache.activemq.artemis.utils.ReusableLatch;
@ -53,9 +53,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.invoke.MethodHandles;
import static org.apache.activemq.artemis.core.server.cluster.qourum.SharedNothingBackupQuorum.BACKUP_ACTIVATION.FAILURE_REPLICATING;
import static org.apache.activemq.artemis.core.server.cluster.qourum.SharedNothingBackupQuorum.BACKUP_ACTIVATION.FAIL_OVER;
import static org.apache.activemq.artemis.core.server.cluster.qourum.SharedNothingBackupQuorum.BACKUP_ACTIVATION.STOP;
import static org.apache.activemq.artemis.core.server.cluster.quorum.SharedNothingBackupQuorum.BACKUP_ACTIVATION.FAILURE_REPLICATING;
import static org.apache.activemq.artemis.core.server.cluster.quorum.SharedNothingBackupQuorum.BACKUP_ACTIVATION.FAIL_OVER;
import static org.apache.activemq.artemis.core.server.cluster.quorum.SharedNothingBackupQuorum.BACKUP_ACTIVATION.STOP;
public final class SharedNothingBackupActivation extends Activation implements ReplicationEndpointEventListener {

View File

@ -58,7 +58,7 @@ import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.server.NodeManager;
import org.apache.activemq.artemis.core.server.cluster.ClusterConnection;
import org.apache.activemq.artemis.core.server.cluster.ha.ReplicatedPolicy;
import org.apache.activemq.artemis.core.server.cluster.qourum.QuorumManager;
import org.apache.activemq.artemis.core.server.cluster.quorum.QuorumManager;
import org.apache.activemq.artemis.spi.core.remoting.Acceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

View File

@ -24,14 +24,14 @@ import java.util.concurrent.TimeoutException;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.core.server.NodeManager;
import org.apache.activemq.artemis.quorum.DistributedLock;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.quorum.MutableLong;
import org.apache.activemq.artemis.quorum.UnavailableStateException;
import org.apache.activemq.artemis.lockmanager.DistributedLock;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
import org.apache.activemq.artemis.lockmanager.MutableLong;
import org.apache.activemq.artemis.lockmanager.UnavailableStateException;
import org.slf4j.Logger;
/**
* This class contains the activation sequence logic of the pluggable quorum vote:
* This class contains the activation sequence logic of the pluggable lock manager:
* it should be used by {@link org.apache.activemq.artemis.core.server.impl.ReplicationBackupActivation}
* and {@link org.apache.activemq.artemis.core.server.impl.ReplicationPrimaryActivation} to coordinate
* for replication.
@ -60,7 +60,7 @@ public final class ActivationSequenceStateMachine {
* the activation and guarantee the initial not-replicated ownership of data.
*/
public static DistributedLock tryActivate(final NodeManager nodeManager,
final DistributedPrimitiveManager distributedManager,
final DistributedLockManager distributedManager,
final Logger logger) throws InterruptedException, ExecutionException, TimeoutException, UnavailableStateException {
Objects.requireNonNull(nodeManager);
Objects.requireNonNull(distributedManager);
@ -199,7 +199,7 @@ public final class ActivationSequenceStateMachine {
/**
* It wait until {@code timeoutMillis ms} has passed or the coordinated activation sequence has progressed enough
*/
public static boolean awaitNextCommittedActivationSequence(final DistributedPrimitiveManager distributedManager,
public static boolean awaitNextCommittedActivationSequence(final DistributedLockManager distributedManager,
final String coordinatedLockAndNodeId,
final long activationSequence,
final long timeoutMills,
@ -262,7 +262,7 @@ public final class ActivationSequenceStateMachine {
*/
public static void ensureSequentialAccessToNodeData(final String serverDescription,
final NodeManager nodeManager,
final DistributedPrimitiveManager distributedManager,
final DistributedLockManager distributedManager,
final Logger logger) throws ActiveMQException, InterruptedException, UnavailableStateException, ExecutionException, TimeoutException {
Objects.requireNonNull(serverDescription);

View File

@ -1135,7 +1135,7 @@ public class ConfigurationImplTest extends ServerTestBase {
@Test
public void testReplicatedPolicyConfiguration() throws Throwable {
Properties properties = new ConfigurationImpl.InsertionOrderedProperties();
properties.put("HAPolicyConfiguration", "REPLICATED");
properties.put("HAPolicyConfiguration", "REPLICATION_PRIMARY_QUORUM_VOTING");
properties.put("HAPolicyConfiguration.checkForActiveServer", "true");
properties.put("HAPolicyConfiguration.groupName", "g0");
properties.put("HAPolicyConfiguration.clusterName", "c0");
@ -1170,7 +1170,7 @@ public class ConfigurationImplTest extends ServerTestBase {
@Test
public void testReplicaPolicyConfiguration() throws Throwable {
Properties properties = new ConfigurationImpl.InsertionOrderedProperties();
properties.put("HAPolicyConfiguration", "REPLICA");
properties.put("HAPolicyConfiguration", "REPLICATION_BACKUP_QUORUM_VOTING");
properties.put("HAPolicyConfiguration.clusterName", "c0");
properties.put("HAPolicyConfiguration.maxSavedReplicatedJournalsSize", "3");
properties.put("HAPolicyConfiguration.groupName", "g0");
@ -1288,7 +1288,7 @@ public class ConfigurationImplTest extends ServerTestBase {
@Test
public void testReplicationPrimaryPolicyConfiguration() throws Throwable {
Properties properties = new ConfigurationImpl.InsertionOrderedProperties();
properties.put("HAPolicyConfiguration", "REPLICATION_PRIMARY");
properties.put("HAPolicyConfiguration", "REPLICATION_PRIMARY_LOCK_MANAGER");
properties.put("HAPolicyConfiguration.groupName", "g0");
properties.put("HAPolicyConfiguration.clusterName", "c0");
properties.put("HAPolicyConfiguration.initialReplicationSyncTimeout", "5");
@ -1319,7 +1319,7 @@ public class ConfigurationImplTest extends ServerTestBase {
@Test
public void testReplicationBackupPolicyConfiguration() throws Throwable {
Properties properties = new ConfigurationImpl.InsertionOrderedProperties();
properties.put("HAPolicyConfiguration", "REPLICATION_BACKUP");
properties.put("HAPolicyConfiguration", "REPLICATION_BACKUP_LOCK_MANAGER");
properties.put("HAPolicyConfiguration.clusterName", "c0");
properties.put("HAPolicyConfiguration.maxSavedReplicatedJournalsSize", "3");
properties.put("HAPolicyConfiguration.groupName", "g0");

View File

@ -46,10 +46,10 @@ import org.apache.activemq.artemis.core.server.impl.SharedNothingBackupActivatio
import org.apache.activemq.artemis.core.server.impl.SharedNothingPrimaryActivation;
import org.apache.activemq.artemis.core.server.impl.SharedStoreBackupActivation;
import org.apache.activemq.artemis.core.server.impl.SharedStorePrimaryActivation;
import org.apache.activemq.artemis.quorum.DistributedLock;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.quorum.MutableLong;
import org.apache.activemq.artemis.quorum.UnavailableStateException;
import org.apache.activemq.artemis.lockmanager.DistributedLock;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
import org.apache.activemq.artemis.lockmanager.MutableLong;
import org.apache.activemq.artemis.lockmanager.UnavailableStateException;
import org.apache.activemq.artemis.tests.util.ServerTestBase;
import org.junit.After;
import org.junit.Test;
@ -142,13 +142,13 @@ public class HAPolicyConfigurationTest extends ServerTestBase {
primaryOnlyTest("primary-only-hapolicy-config5.xml");
}
public static class FakeDistributedPrimitiveManager implements DistributedPrimitiveManager {
public static class FakeDistributedLockManager implements DistributedLockManager {
private final Map<String, String> config;
private boolean started;
private DistributedLock lock;
public FakeDistributedPrimitiveManager(Map<String, String> config) {
public FakeDistributedLockManager(Map<String, String> config) {
this.config = config;
this.started = false;
}
@ -333,11 +333,11 @@ public class HAPolicyConfigurationTest extends ServerTestBase {
assertNull(failbackPolicy.getScaleDownClustername());
assertNull(failbackPolicy.getScaleDownGroupName());
// validate manager
DistributedPrimitiveManager manager = ((ReplicationPrimaryActivation) activation).getDistributedManager();
DistributedLockManager manager = ((ReplicationPrimaryActivation) activation).getDistributedManager();
assertNotNull(manager);
assertEquals(FakeDistributedPrimitiveManager.class.getName(), manager.getClass().getName());
assertTrue(manager + " is not an instance of FakeDistributedPrimitiveManager", manager instanceof FakeDistributedPrimitiveManager);
FakeDistributedPrimitiveManager forwardingManager = (FakeDistributedPrimitiveManager) manager;
assertEquals(FakeDistributedLockManager.class.getName(), manager.getClass().getName());
assertTrue(manager + " is not an instance of FakeDistributedLockManager", manager instanceof FakeDistributedLockManager);
FakeDistributedLockManager forwardingManager = (FakeDistributedLockManager) manager;
// validate manager config
validateManagerConfig(forwardingManager.getConfig());
} finally {
@ -389,11 +389,11 @@ public class HAPolicyConfigurationTest extends ServerTestBase {
assertNull(failoverPrimaryPolicy.getScaleDownClustername());
assertNull(failoverPrimaryPolicy.getScaleDownGroupName());
// validate manager
DistributedPrimitiveManager manager = ((ReplicationBackupActivation) activation).getDistributedManager();
DistributedLockManager manager = ((ReplicationBackupActivation) activation).getDistributedManager();
assertNotNull(manager);
assertEquals(FakeDistributedPrimitiveManager.class.getName(), manager.getClass().getName());
assertTrue(manager instanceof FakeDistributedPrimitiveManager);
FakeDistributedPrimitiveManager forwardingManager = (FakeDistributedPrimitiveManager) manager;
assertEquals(FakeDistributedLockManager.class.getName(), manager.getClass().getName());
assertTrue(manager instanceof FakeDistributedLockManager);
FakeDistributedLockManager forwardingManager = (FakeDistributedLockManager) manager;
// validate manager config
validateManagerConfig(forwardingManager.getConfig());
} finally {

View File

@ -54,11 +54,11 @@ import org.apache.activemq.artemis.core.server.cluster.ha.HAPolicy;
import org.apache.activemq.artemis.core.server.cluster.ha.ReplicaPolicy;
import org.apache.activemq.artemis.core.server.cluster.ha.ReplicationBackupPolicy;
import org.apache.activemq.artemis.core.server.cluster.ha.ReplicationPrimaryPolicy;
import org.apache.activemq.artemis.core.server.cluster.qourum.QuorumManager;
import org.apache.activemq.artemis.core.server.cluster.qourum.SharedNothingBackupQuorum;
import org.apache.activemq.artemis.core.server.cluster.quorum.QuorumManager;
import org.apache.activemq.artemis.core.server.cluster.quorum.SharedNothingBackupQuorum;
import org.apache.activemq.artemis.core.server.management.ManagementService;
import org.apache.activemq.artemis.core.version.Version;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
import org.apache.activemq.artemis.spi.core.remoting.ClientProtocolManager;
import org.apache.activemq.artemis.utils.ExecutorFactory;
import org.apache.activemq.artemis.utils.ThreadDumpUtil;
@ -86,7 +86,7 @@ public class BackupActivationNoReconnectTest {
when(policy.getPrimaryPolicy()).thenReturn(replicationPrimaryPolicy);
ActiveMQServerImpl server = Mockito.mock(ActiveMQServerImpl.class);
DistributedPrimitiveManager distributedManager = Mockito.mock(DistributedPrimitiveManager.class);
DistributedLockManager distributedManager = Mockito.mock(DistributedLockManager.class);
ReplicationBackupActivation replicationBackupActivation = new ReplicationBackupActivation(server, distributedManager, policy);
verifySingleAttemptToLocatePrimary(server, replicationBackupActivation);

View File

@ -33,7 +33,7 @@
<allow-failback>false</allow-failback>
<manager>
<class-name>
org.apache.activemq.artemis.core.config.impl.HAPolicyConfigurationTest$FakeDistributedPrimitiveManager
org.apache.activemq.artemis.core.config.impl.HAPolicyConfigurationTest$FakeDistributedLockManager
</class-name>
<properties>
<property key="connect-string" value="127.0.0.1:6666"/>

View File

@ -30,7 +30,7 @@
<max-saved-replicated-journals-size>73</max-saved-replicated-journals-size>
<manager>
<class-name>
org.apache.activemq.artemis.core.config.impl.HAPolicyConfigurationTest$FakeDistributedPrimitiveManager
org.apache.activemq.artemis.core.config.impl.HAPolicyConfigurationTest$FakeDistributedLockManager
</class-name>
<properties>
<property key="connect-string" value="127.0.0.1:6666"/>

View File

@ -2,7 +2,7 @@
:idprefix:
:idseparator: -
You can use the Artemis CLI to execute activation sequence maintenance/recovery tools for xref:ha.adoc#high-availability-and-failover[Pluggable Quorum Replication].
You can use the Artemis CLI to execute activation sequence maintenance/recovery tools for xref:ha.adoc#replication[Replication] with Pluggable Lock Manager.
The 2 main commands are `activation list` and `activation set`, that can be used together to recover some disaster happened to local/coordinated activation sequences.

File diff suppressed because it is too large Load Diff

View File

@ -2,110 +2,103 @@
:idprefix:
:idseparator: -
It is possible that if a primary or backup configured for replication becomes isolated in a network that failover will occur and you will end up with 2 active brokers serving messages. This we call _split brain_.
There are different configurations you can choose from that will help mitigate this problem.
A _split brain_ is a condition that occurs when two different brokers are serving the same messages at the same time.
When this happens instead of client applications all sharing the _same_ broker as they ought, they may become divided between the two split brain brokers.
This is problematic because it can lead to:
* *Duplicate messages* e.g. when multiple consumers on the same JMS queue split between both brokers and receive the same message(s)
* *Missed messages* e.g. when multiple consumers on the same JMS topic split between both brokers and producers are only sending messages to one broker
Split brain most commonly happens when a pair of brokers in an HA *replication* configuration lose the replication connection linking them together.
When this connection is lost the backup assumes that the primary has died and therefore activates.
At this point there are two brokers on the network which are isolated from each other and since the backup has a copy of all the messages from the primary they are each serving the same messages.
[IMPORTANT]
.What about shared store configurations?
====
While it is technically possible for split brain to happen with a pair of brokers in an HA _shared store_ configuration it would require a failure in the file-locking mechanism of the storage device which the brokers are sharing.
One of the benefits of using a shared store is that the storage device itself acts as an arbiter to ensure consistency and mitigate split brain.
====
Recovering from a split brain may be as simple as stopping the broker which activated by mistake.
However, this solution is only viable *if* no client application connected to it and performed messaging operations.
The longer client applications are allowed to interact with split brain brokers the more difficult it will be to understand and remediate the resulting problems.
There are several different configurations you can choose from that will help mitigate split brain.
== Pluggable Lock Manager
A pluggable lock manager configuration requires a 3rd party to establish a shared lock between primary and backup brokers.
The shared lock ensures that either the primary or backup is active at any given point in time, similar to how the file lock functions in the shared storage use-case.
The _plugin_ decides what 3rd party implementation is used.
It could be something as simple as a shared file on a network file system that supports locking (e.g. NFS) or it could be something more complex like https://etcd.io/[etcd].
The broker ships with a xref:ha.adoc#apache-zookeeper-integration[reference plugin implementation] based on https://zookeeper.apache.org/[Apache ZooKeeper] - a common implementation used for this kind of task.
The main benefit of a pluggable lock manager is that is releases the broker from the responsibility of establishing a reliable vote.
This means that a _single_ HA pair of brokers can be reliably protected against split-brain.
== Quorum Voting
Quorum voting is used by both the primary and the backup to decide what to do if a replication connection is disconnected.
Basically the server will request each active server in the cluster to vote as to whether it thinks the server it is replicating to or from is still alive.
You can also configure the time for which the quorum manager will wait for the quorum vote response.
The default time is 30 seconds you can configure like so for primary and also for the backup:
Quorum voting is a process by which one node in a cluster can determine whether another node in the cluster is active without directly communicating with that node.
Then the broker initiating the vote can take action based on the result (e.g. shutting itself down to avoid split-brain).
[,xml]
----
<ha-policy>
<replication>
<primary>
<quorum-vote-wait>12</quorum-vote-wait>
</primary>
</replication>
</ha-policy>
----
Quorum voting requires the participation of the other _active_ brokers in the cluster.
Of course this requires that there are, in fact, other active brokers in the cluster which means quorum voting won't work with a single HA pair of brokers.
Furthermore, it also won't work with just two HA pairs of brokers either because that's still not enough for a legitimate quorum.
There must be at least three HA pairs to establish a proper quorum with quorum voting.
This being the case the minimum number of live/backup pairs needed is 3.
If less than 3 pairs are used then the only option is to use a Network Pinger which is explained later in this chapter or choose how you want each server to react which the following details:
=== Voting Mechanics
=== Backup Voting
When the replication connection between a primary and backup is lost the backup and/or the primary may initiate a vote.
By default if a backup loses its replication connection to its primary it makes a decision as to whether to start or not with a quorum vote.
This of course requires that there be at least 3 pairs of primary/backup nodes in the cluster.
For a 3 node cluster it will start if it gets 2 votes back saying that its primary server is no longer available, for 4 nodes this would be 3 votes and so on.
When a backup loses connection to the primary it will keep voting for a quorum until it either receives a vote allowing it to start or it detects that the primary is still active.
for the latter it will then restart as a backup.
How many votes and how long between each vote the backup should wait is configured like so:
[IMPORTANT]
====
For a vote to pass a _majority_ of affirmative responses is required.
For example, in a 3 node cluster a vote will pass with 2 affirmatives.
For a 4 node cluster this would be 3 affirmatives and so on.
====
[,xml]
----
<ha-policy>
<replication>
<backup>
<vote-retries>12</vote-retries>
<vote-retry-wait>5000</vote-retry-wait>
</backup>
</replication>
</ha-policy>
----
==== Backup Voting
It's also possible to statically set the quorum size that should be used for the case where the cluster size is known up front, this is done on the Replica Policy like so:
By default, if a backup loses its replication connection to its primary it will activate automatically.
However, it can be configured via the `vote-on-replication-failure` property to initiate a quorum vote in order to decide whether to activate or not.
If this is done then the backup will keep voting until it either receives a vote allowing it to start or it detects that the primary is still active.
In the latter case it will then restart as a backup.
[,xml]
----
<ha-policy>
<replication>
<backup>
<quorum-size>2</quorum-size>
</backup>
</replication>
</ha-policy>
----
See the section on xref:ha.adoc#replication-configuration[Replication Configuration] for more details on configuration.
In this example the quorum size is set to 2 so if you were using a single pair and the backup lost connectivity it would never start.
==== Primary Voting
=== Primary Voting
By default, if the primary server loses its replication connection to the backup then it will just carry on and wait for a backup to reconnect and start replicating again.
However, this may mean that the primary remains active even though the backup has activated so this behavior is configurable via the `vote-on-replication-failure` property.
By default, if the primary server loses its replication connection then it will just carry on and wait for a backup to reconnect and start replicating again.
In the event of a possible split brain scenario this may mean that the primary stays active even though the backup has been activated.
It is possible to configure the primary server to vote for a quorum if this happens, in this way if the primary server does not receive a majority vote then it will shutdown.
This is done by setting the _vote-on-replication-failure_ to `true`.
[,xml]
----
<ha-policy>
<replication>
<primary>
<vote-on-replication-failure>true</vote-on-replication-failure>
<quorum-size>2</quorum-size>
</primary>
</replication>
</ha-policy>
----
As in the backup policy it is also possible to statically configure the quorum size.
See the section on xref:ha.adoc#replication-configuration[Replication Configuration] for more details on configuration.
== Pinging the network
You may configure one more addresses on the broker.xml that are part of your network topology, that will be pinged through the life cycle of the server.
You may configure one more addresses in `broker.xml` that that will be pinged throughout the life of the server. The server will stop itself if it can't ping one or more of the addresses in the list.
The server will stop itself until the network is back on such case.
If you execute the create command passing a -ping argument, you will create a default xml that is ready to be used with network checks:
If you execute the `create` command using the `--ping` argument you will create a default XML that is ready to be used with network checks:
[,console]
----
./artemis create /myDir/myServer --ping 10.0.0.1
$ ./artemis create /myDir/myServer --ping 10.0.0.1
----
This XML part will be added to your broker.xml:
This XML will be added to your `broker.xml`:
[,xml]
----
<!--
You can verify the network health of a particular NIC by specifying the <network-check-NIC> element.
You can verify the network health of a particular NIC by specifying the <network-check-NIC> element.
<network-check-NIC>theNicName</network-check-NIC>
-->
-->
<!--
Use this to use an HTTP server to validate the network
Use this to use an HTTP server to validate the network
<network-check-URL-list>http://www.apache.org</network-check-URL-list> -->
<network-check-period>10000</network-check-period>
@ -122,12 +115,10 @@ Use this to use an HTTP server to validate the network
<!-- use this to customize the ping used for ipv4 addresses -->
<network-check-ping-command>ping -c 1 -t %d %s</network-check-ping-command>
<!-- use this to customize the ping used for ipv addresses -->
<!-- use this to customize the ping used for ipv6 addresses -->
<network-check-ping6-command>ping6 -c 1 %2$s</network-check-ping6-command>
----
Once you lose connectivity towards 10.0.0.1 on the given example, you will see see this output at the server:
Once you lose connectivity towards `10.0.0.1` on the given example the broker will log something like this:
----
09:49:24,562 WARN [org.apache.activemq.artemis.core.server.NetworkHealthCheck] Ping Address /10.0.0.1 wasn't reacheable
09:49:36,577 INFO [org.apache.activemq.artemis.core.server.NetworkHealthCheck] Network is unhealthy, stopping service ActiveMQServerImpl::serverUUID=04fd5dd8-b18c-11e6-9efe-6a0001921ad0
@ -151,12 +142,11 @@ Once you lose connectivity towards 10.0.0.1 on the given example, you will see s
at java.lang.Thread.run(Thread.java:745) [rt.jar:1.8.0_73]
----
Once you re establish your network connections towards the configured check list:
Once you reestablish your network connections towards the configured check-list:
----
09:53:23,461 INFO [org.apache.activemq.artemis.core.server.NetworkHealthCheck] Network is healthy, starting service ActiveMQServerImpl::
09:53:23,462 INFO [org.apache.activemq.artemis.core.server] AMQ221000: live Message Broker is starting with configuration Broker Configuration (clustered=false,journalDirectory=./data/journal,bindingsDirectory=./data/bindings,largeMessagesDirectory=./data/large-messages,pagingDirectory=./data/paging)
09:53:23,462 INFO [org.apache.activemq.artemis.core.server] AMQ221000: primary Message Broker is starting with configuration Broker Configuration (clustered=false,journalDirectory=./data/journal,bindingsDirectory=./data/bindings,largeMessagesDirectory=./data/large-messages,pagingDirectory=./data/paging)
09:53:23,462 INFO [org.apache.activemq.artemis.core.server] AMQ221013: Using NIO Journal
09:53:23,462 INFO [org.apache.activemq.artemis.core.server] AMQ221043: Protocol module found: [artemis-server]. Adding protocol support for: CORE
09:53:23,463 INFO [org.apache.activemq.artemis.core.server] AMQ221043: Protocol module found: [artemis-amqp-protocol]. Adding protocol support for: AMQP
@ -171,11 +161,11 @@ Once you re establish your network connections towards the configured check list
09:53:23,554 INFO [org.apache.activemq.artemis.core.server] AMQ221020: Started Acceptor at 0.0.0.0:5672 for protocols [AMQP]
09:53:23,555 INFO [org.apache.activemq.artemis.core.server] AMQ221020: Started Acceptor at 0.0.0.0:1883 for protocols [MQTT]
09:53:23,556 INFO [org.apache.activemq.artemis.core.server] AMQ221020: Started Acceptor at 0.0.0.0:61613 for protocols [STOMP]
09:53:23,556 INFO [org.apache.activemq.artemis.core.server] AMQ221007: Server is now live
09:53:23,556 INFO [org.apache.activemq.artemis.core.server] AMQ221007: Server is now active
09:53:23,556 INFO [org.apache.activemq.artemis.core.server] AMQ221001: Apache ActiveMQ Artemis Message Broker version 1.6.0 [0.0.0.0, nodeID=04fd5dd8-b18c-11e6-9efe-6a0001921ad0]
----
[WARNING]
[IMPORTANT]
====
Make sure you understand your network topology as this is meant to validate your network.
Using IPs that could eventually disappear or be partially visible may defeat the purpose.

View File

@ -33,6 +33,7 @@ The dot (`.`) character here is part of the broker's wildcard syntax, and it is
+
In this case the characters from the broker's wildcard syntax that do not match the characters in the MQTT wildcard syntax will be escaped with a backslash (i.e. `\`).
To avoid this conversion you can configure the broker to use the MQTT wildcard syntax or change the name of the MQTT topic name or filter.
* Due to https://issues.apache.org/jira/browse/ARTEMIS-4559[ARTEMIS-4559] folks embedding the broker and also depending on the `artemis-quorum-ri` and/or `artemis-quorum-api` modules and/or using `org.apache.activemq.artemis.core.config.ha.DistributedPrimitiveManagerConfiguration` will need to use `artemis-lockmanager-ri`, `artemis-lockmanager-api`, and `org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration` respectively. Previously these were marked as "experimental" in the documentation and were changed strictly in name to clarify their use conceptually. Furthermore, the documentation around high availability and network isolation (i.e. split brain) was refactored significantly to be more clear and comprehensive.
== 2.32.0

View File

@ -66,8 +66,7 @@
<module>artemis-unit-test-support</module>
<module>tests</module>
<module>artemis-features</module>
<module>artemis-quorum-api</module>
<module>artemis-quorum-ri</module>
<module>artemis-lockmanager</module>
<module>artemis-image</module>
<module>artemis-image/examples</module>
</modules>

View File

@ -218,7 +218,7 @@ under the License.
<!--
class-name defaults to:
<class-name>
org.apache.activemq.artemis.core.config.impl.HAPolicyConfigurationTest$FakeDistributedPrimitiveManager
org.apache.activemq.artemis.core.config.impl.HAPolicyConfigurationTest$FakeDistributedLockManager
</class-name>
-->
<properties>

View File

@ -216,7 +216,7 @@ under the License.
<!--
class-name defaults to:
<class-name>
org.apache.activemq.artemis.core.config.impl.HAPolicyConfigurationTest$FakeDistributedPrimitiveManager
org.apache.activemq.artemis.core.config.impl.HAPolicyConfigurationTest$FakeDistributedLockManager
</class-name>
-->
<properties>

View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.e2e.ha.replication.pluggablequorumvote;
package org.apache.activemq.artemis.tests.e2e.ha.replication.lockmanager;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
@ -41,7 +41,7 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
public class GCPauseSimulationTests extends PluggableQuorumVoteTestBase {
public class GCPauseSimulationTests extends LockManagerTestBase {
private static final String PRIMARY_LOCATION = E2ETestBase.basedir + "/target/ha/replication/pluggablequorumvote/primary";
private static final String BACKUP_LOCATION = E2ETestBase.basedir + "/target/ha/replication/pluggablequorumvote/backup";

View File

@ -15,13 +15,13 @@
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.e2e.ha.replication.pluggablequorumvote;
package org.apache.activemq.artemis.tests.e2e.ha.replication.lockmanager;
import org.apache.activemq.artemis.tests.e2e.common.ContainerService;
import org.junit.AfterClass;
import org.junit.BeforeClass;
public abstract class PluggableQuorumVoteTestBase {
public abstract class LockManagerTestBase {
protected static ContainerService service;
protected static Object network;

View File

@ -35,12 +35,12 @@
<dependencies>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-quorum-api</artifactId>
<artifactId>artemis-lockmanager-api</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-quorum-ri</artifactId>
<artifactId>artemis-lockmanager-ri</artifactId>
<scope>test</scope>
</dependency>
<dependency>

View File

@ -26,17 +26,17 @@ import org.apache.activemq.artemis.cli.commands.activation.ActivationSequenceSet
import org.apache.activemq.artemis.core.config.ClusterConnectionConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.HAPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedPrimitiveManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationPrimaryPolicyConfiguration;
import org.apache.activemq.artemis.core.config.impl.ConfigurationImpl;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.ActiveMQServers;
import org.apache.activemq.artemis.core.server.JournalType;
import org.apache.activemq.artemis.quorum.DistributedLock;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.quorum.MutableLong;
import org.apache.activemq.artemis.quorum.file.FileBasedPrimitiveManager;
import org.apache.activemq.artemis.lockmanager.DistributedLock;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
import org.apache.activemq.artemis.lockmanager.MutableLong;
import org.apache.activemq.artemis.lockmanager.file.FileBasedLockManager;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.tests.util.Wait;
import org.junit.After;
@ -46,18 +46,20 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import static org.apache.activemq.artemis.lockmanager.DistributedLockManager.newInstanceOf;
public class ActivationSequenceCommandsTest extends ActiveMQTestBase {
@Rule
public TemporaryFolder brokersFolder = new TemporaryFolder();
protected DistributedPrimitiveManagerConfiguration managerConfiguration;
protected DistributedLockManagerConfiguration managerConfiguration;
@Before
@Override
public void setUp() throws Exception {
super.setUp();
managerConfiguration = new DistributedPrimitiveManagerConfiguration(FileBasedPrimitiveManager.class.getName(),
managerConfiguration = new DistributedLockManagerConfiguration(FileBasedLockManager.class.getName(),
Collections.singletonMap("locks-folder", temporaryFolder.newFolder("manager").toString()));
}
@ -209,11 +211,10 @@ public class ActivationSequenceCommandsTest extends ActiveMQTestBase {
ActivationSequenceList.ListResult list = ActivationSequenceList.execute(sequenceList, primaryConfiguration, null);
Assert.assertEquals(expectedStartCoordinatedSequence, list.coordinatedActivationSequence.longValue());
Assert.assertEquals(expectedStartCoordinatedSequence, list.localActivationSequence.longValue());
try (DistributedPrimitiveManager distributedPrimitiveManager = DistributedPrimitiveManager
.newInstanceOf(managerConfiguration.getClassName(), managerConfiguration.getProperties())) {
distributedPrimitiveManager.start();
try (DistributedLock lock = distributedPrimitiveManager.getDistributedLock(nodeID);
MutableLong coordinatedActivationSequence = distributedPrimitiveManager.getMutableLong(nodeID)) {
try (DistributedLockManager DistributedLockManager = newInstanceOf(managerConfiguration.getClassName(), managerConfiguration.getProperties())) {
DistributedLockManager.start();
try (DistributedLock lock = DistributedLockManager.getDistributedLock(nodeID);
MutableLong coordinatedActivationSequence = DistributedLockManager.getMutableLong(nodeID)) {
Assert.assertTrue(lock.tryLock());
final long activationSequence = coordinatedActivationSequence.get();
Assert.assertEquals(expectedStartCoordinatedSequence, activationSequence);

View File

@ -19,17 +19,17 @@ package org.apache.activemq.artemis.tests.integration.client;
import java.util.Collections;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedPrimitiveManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.quorum.file.FileBasedPrimitiveManager;
import org.apache.activemq.artemis.lockmanager.file.FileBasedLockManager;
import org.apache.activemq.artemis.tests.util.ReplicatedBackupUtils;
import org.junit.Before;
public class PluggableQuorumInfiniteRedeliveryTest extends InfiniteRedeliveryTest {
public class LockManagerInfiniteRedeliveryTest extends InfiniteRedeliveryTest {
private DistributedPrimitiveManagerConfiguration managerConfiguration;
private DistributedLockManagerConfiguration managerConfiguration;
public PluggableQuorumInfiniteRedeliveryTest(String protocol, boolean useCLI) {
public LockManagerInfiniteRedeliveryTest(String protocol, boolean useCLI) {
super(protocol, useCLI);
}
@ -37,7 +37,7 @@ public class PluggableQuorumInfiniteRedeliveryTest extends InfiniteRedeliveryTes
@Override
public void setUp() throws Exception {
super.setUp();
this.managerConfiguration = new DistributedPrimitiveManagerConfiguration(FileBasedPrimitiveManager.class.getName(),
this.managerConfiguration = new DistributedLockManagerConfiguration(FileBasedLockManager.class.getName(),
Collections.singletonMap("locks-folder", temporaryFolder.newFolder("manager").toString()));
}

View File

@ -57,7 +57,7 @@ import org.apache.activemq.artemis.core.client.impl.TopologyMemberImpl;
import org.apache.activemq.artemis.core.config.ClusterConnectionConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.HAPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedPrimitiveManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.PrimaryOnlyPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicaPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicatedPolicyConfiguration;
@ -84,12 +84,12 @@ import org.apache.activemq.artemis.core.server.cluster.impl.BridgeMetrics;
import org.apache.activemq.artemis.core.server.cluster.impl.ClusterConnectionImpl;
import org.apache.activemq.artemis.core.server.cluster.impl.ClusterConnectionMetrics;
import org.apache.activemq.artemis.core.server.cluster.impl.MessageLoadBalancingType;
import org.apache.activemq.artemis.core.server.cluster.qourum.SharedNothingBackupQuorum;
import org.apache.activemq.artemis.core.server.cluster.quorum.SharedNothingBackupQuorum;
import org.apache.activemq.artemis.core.server.group.GroupingHandler;
import org.apache.activemq.artemis.core.server.group.impl.GroupingHandlerConfiguration;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.server.impl.InVMNodeManager;
import org.apache.activemq.artemis.quorum.file.FileBasedPrimitiveManager;
import org.apache.activemq.artemis.lockmanager.file.FileBasedLockManager;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.utils.PortCheckRule;
import org.junit.After;
@ -141,14 +141,14 @@ public abstract class ClusterTestBase extends ActiveMQTestBase {
return true;
}
private DistributedPrimitiveManagerConfiguration pluggableQuorumConfiguration = null;
private DistributedLockManagerConfiguration pluggableQuorumConfiguration = null;
private DistributedPrimitiveManagerConfiguration getOrCreatePluggableQuorumConfiguration() {
private DistributedLockManagerConfiguration getOrCreatePluggableQuorumConfiguration() {
if (pluggableQuorumConfiguration != null) {
return pluggableQuorumConfiguration;
}
try {
pluggableQuorumConfiguration = new DistributedPrimitiveManagerConfiguration(FileBasedPrimitiveManager.class.getName(), Collections.singletonMap("locks-folder", temporaryFolder.newFolder("manager").toString()));
pluggableQuorumConfiguration = new DistributedLockManagerConfiguration(FileBasedLockManager.class.getName(), Collections.singletonMap("locks-folder", temporaryFolder.newFolder("manager").toString()));
} catch (IOException ioException) {
logger.error(ioException.getMessage(), ioException);
return null;

View File

@ -37,7 +37,7 @@ import org.apache.activemq.artemis.core.client.impl.ClientSessionFactoryInternal
import org.apache.activemq.artemis.core.client.impl.ServerLocatorInternal;
import org.apache.activemq.artemis.core.config.ClusterConnectionConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.ha.DistributedPrimitiveManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicaPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.SharedStorePrimaryPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.SharedStoreBackupPolicyConfiguration;
@ -48,7 +48,7 @@ import org.apache.activemq.artemis.core.server.cluster.ha.HAPolicy;
import org.apache.activemq.artemis.core.server.cluster.ha.ReplicatedPolicy;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.core.server.impl.InVMNodeManager;
import org.apache.activemq.artemis.quorum.file.FileBasedPrimitiveManager;
import org.apache.activemq.artemis.lockmanager.file.FileBasedLockManager;
import org.apache.activemq.artemis.tests.integration.cluster.util.SameProcessActiveMQServer;
import org.apache.activemq.artemis.tests.integration.cluster.util.TestableServer;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
@ -83,7 +83,7 @@ public abstract class FailoverTestBase extends ActiveMQTestBase {
protected NodeManager backupNodeManager;
protected DistributedPrimitiveManagerConfiguration managerConfiguration;
protected DistributedLockManagerConfiguration managerConfiguration;
protected boolean startBackupServer = true;
@ -236,7 +236,7 @@ public abstract class FailoverTestBase extends ActiveMQTestBase {
primaryConfig = createDefaultInVMConfig();
managerConfiguration =
new DistributedPrimitiveManagerConfiguration(FileBasedPrimitiveManager.class.getName(),
new DistributedLockManagerConfiguration(FileBasedLockManager.class.getName(),
Collections.singletonMap("locks-folder", temporaryFolder.newFolder("manager").toString()));
ReplicatedBackupUtils.configurePluggableQuorumReplicationPair(backupConfig, backupConnector, backupAcceptor, primaryConfig, primaryConnector, null, managerConfiguration, managerConfiguration);

View File

@ -29,7 +29,7 @@ import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
import org.apache.activemq.artemis.core.client.impl.ServerLocatorInternal;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.HAPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedPrimitiveManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicaPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicatedPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
@ -39,7 +39,7 @@ import org.apache.activemq.artemis.core.config.ha.SharedStoreBackupPolicyConfigu
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.NodeManager;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.quorum.file.FileBasedPrimitiveManager;
import org.apache.activemq.artemis.lockmanager.file.FileBasedLockManager;
import org.apache.activemq.artemis.tests.integration.cluster.distribution.ClusterTestBase;
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.tests.integration.cluster.util.SameProcessActiveMQServer;
@ -54,14 +54,14 @@ import java.lang.invoke.MethodHandles;
public abstract class MultipleServerFailoverTestBase extends ActiveMQTestBase {
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private DistributedPrimitiveManagerConfiguration pluggableQuorumConfiguration = null;
private DistributedLockManagerConfiguration pluggableQuorumConfiguration = null;
private DistributedPrimitiveManagerConfiguration getOrCreatePluggableQuorumConfiguration() {
private DistributedLockManagerConfiguration getOrCreatePluggableQuorumConfiguration() {
if (pluggableQuorumConfiguration != null) {
return pluggableQuorumConfiguration;
}
try {
pluggableQuorumConfiguration = new DistributedPrimitiveManagerConfiguration(FileBasedPrimitiveManager.class.getName(), Collections.singletonMap("locks-folder", temporaryFolder.newFolder("manager").toString()));
pluggableQuorumConfiguration = new DistributedLockManagerConfiguration(FileBasedLockManager.class.getName(), Collections.singletonMap("locks-folder", temporaryFolder.newFolder("manager").toString()));
} catch (IOException ioException) {
return null;
}

View File

@ -26,8 +26,8 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.activemq.artemis.core.server.cluster.qourum.QuorumVoteServerConnect;
import org.apache.activemq.artemis.core.server.cluster.qourum.ServerConnectVote;
import org.apache.activemq.artemis.core.server.cluster.quorum.QuorumVoteServerConnect;
import org.apache.activemq.artemis.core.server.cluster.quorum.ServerConnectVote;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.hamcrest.Matchers;
import org.junit.Assert;

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.integration.cluster.failover.quorum;
package org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager;
import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
@ -39,7 +39,7 @@ import org.junit.runners.Parameterized;
import static java.util.Arrays.asList;
@RunWith(Parameterized.class)
public class PluggableQuorumBackupAuthenticationTest extends FailoverTestBase {
public class LockManagerBackupAuthenticationTest extends FailoverTestBase {
private static CountDownLatch registrationStarted;

View File

@ -15,12 +15,12 @@
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.integration.cluster.failover.quorum;
package org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.tests.integration.cluster.failover.BackupSyncJournalTest;
public class PluggableQuorumBackupSyncJournalTest extends BackupSyncJournalTest {
public class LockManagerBackupSyncJournalTest extends BackupSyncJournalTest {
@Override
protected void createConfigs() throws Exception {

View File

@ -14,12 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.integration.cluster.failover.quorum;
package org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.tests.integration.cluster.failover.BackupSyncLargeMessageTest;
public class PluggableQuorumBackupSyncLargeMessageTest extends BackupSyncLargeMessageTest {
public class LockManagerBackupSyncLargeMessageTest extends BackupSyncLargeMessageTest {
@Override
protected void createConfigs() throws Exception {

View File

@ -14,12 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.integration.cluster.failover.quorum;
package org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.tests.integration.cluster.failover.BackupSyncPagingTest;
public class PluggableQuorumBackupSyncPagingTest extends BackupSyncPagingTest {
public class LockManagerBackupSyncPagingTest extends BackupSyncPagingTest {
@Override
protected void createConfigs() throws Exception {

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.integration.cluster.failover.quorum;
package org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager;
import java.util.Arrays;
@ -33,7 +33,7 @@ import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@RunWith(Parameterized.class)
public class PluggableQuorumExtraBackupReplicatedFailoverTest extends FailoverTestBase {
public class LockManagerExtraBackupReplicatedFailoverTest extends FailoverTestBase {
private static final String GROUP_NAME = "foo";

View File

@ -14,12 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.integration.cluster.failover.quorum;
package org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager;
import org.apache.activemq.artemis.tests.integration.cluster.distribution.ClusterTestBase;
import org.apache.activemq.artemis.tests.integration.cluster.failover.GroupingFailoverTestBase;
public class PluggableQuorumGroupingFailoverReplicationTest extends GroupingFailoverTestBase {
public class LockManagerGroupingFailoverReplicationTest extends GroupingFailoverTestBase {
@Override
protected ClusterTestBase.HAType haType() {

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.integration.cluster.failover.quorum;
package org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager;
import java.io.IOException;
import java.io.OutputStream;
@ -30,7 +30,7 @@ import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.api.core.client.ClientSession;
import org.apache.activemq.artemis.component.WebServerComponent;
import org.apache.activemq.artemis.core.config.ha.DistributedPrimitiveManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.NodeManager;
@ -38,8 +38,8 @@ import org.apache.activemq.artemis.core.server.ServiceComponent;
import org.apache.activemq.artemis.dto.AppDTO;
import org.apache.activemq.artemis.dto.BindingDTO;
import org.apache.activemq.artemis.dto.WebServerDTO;
import org.apache.activemq.artemis.quorum.MutableLong;
import org.apache.activemq.artemis.quorum.file.FileBasedPrimitiveManager;
import org.apache.activemq.artemis.lockmanager.MutableLong;
import org.apache.activemq.artemis.lockmanager.file.FileBasedLockManager;
import org.apache.activemq.artemis.tests.integration.cluster.failover.FailoverTest;
import org.apache.activemq.artemis.tests.integration.cluster.util.TestableServer;
import org.apache.activemq.artemis.tests.util.Wait;
@ -49,7 +49,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.invoke.MethodHandles;
public class PluggableQuorumNettyNoGroupNameReplicatedFailoverTest extends FailoverTest {
public class LockManagerNettyNoGroupNameReplicatedFailoverTest extends FailoverTest {
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
protected void beforeWaitForRemoteBackupSynchronization() {
@ -225,11 +225,11 @@ public class PluggableQuorumNettyNoGroupNameReplicatedFailoverTest extends Failo
doDecrementActivationSequenceForForceRestartOf(logger, nodeManager, managerConfiguration);
}
public static void doDecrementActivationSequenceForForceRestartOf(Logger log, NodeManager nodeManager, DistributedPrimitiveManagerConfiguration distributedPrimitiveManagerConfiguration) throws Exception {
public static void doDecrementActivationSequenceForForceRestartOf(Logger log, NodeManager nodeManager, DistributedLockManagerConfiguration DistributedLockManagerConfiguration) throws Exception {
nodeManager.start();
long localActivation = nodeManager.readNodeActivationSequence();
// file based
FileBasedPrimitiveManager fileBasedPrimitiveManager = new FileBasedPrimitiveManager(distributedPrimitiveManagerConfiguration.getProperties());
FileBasedLockManager fileBasedPrimitiveManager = new FileBasedLockManager(DistributedLockManagerConfiguration.getProperties());
fileBasedPrimitiveManager.start();
try {
MutableLong mutableLong = fileBasedPrimitiveManager.getMutableLong(nodeManager.getNodeId().toString());

View File

@ -14,12 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.integration.cluster.failover.quorum;
package org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.tests.integration.cluster.failover.NettyReplicationStopTest;
public class PluggableQuorumNettyReplicationStopTest extends NettyReplicationStopTest {
public class LockManagerNettyReplicationStopTest extends NettyReplicationStopTest {
@Override
protected void createConfigs() throws Exception {

View File

@ -14,12 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.integration.cluster.failover.quorum;
package org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.tests.integration.cluster.failover.PageCleanupWhileReplicaCatchupTest;
public class PluggableQuorumPageCleanupWhileReplicaCatchupTest extends PageCleanupWhileReplicaCatchupTest {
public class LockManagerPageCleanupWhileReplicaCatchupTest extends PageCleanupWhileReplicaCatchupTest {
@Override
protected void createConfigs() throws Exception {

View File

@ -14,21 +14,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.integration.cluster.failover.quorum;
package org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager;
import java.io.IOException;
import java.util.Collections;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.ha.DistributedPrimitiveManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationPrimaryPolicyConfiguration;
import org.apache.activemq.artemis.quorum.file.FileBasedPrimitiveManager;
import org.apache.activemq.artemis.lockmanager.file.FileBasedLockManager;
import org.apache.activemq.artemis.tests.integration.cluster.failover.ReplicaTimeoutTest;
import org.apache.activemq.artemis.tests.util.ReplicatedBackupUtils;
public class PluggableQuorumReplicaTimeoutTest extends ReplicaTimeoutTest {
public class LockManagerReplicaTimeoutTest extends ReplicaTimeoutTest {
@Override
protected void configureReplicationPair(Configuration backupConfig,
@ -36,7 +36,7 @@ public class PluggableQuorumReplicaTimeoutTest extends ReplicaTimeoutTest {
TransportConfiguration backupConnector,
TransportConfiguration backupAcceptor,
TransportConfiguration primaryConnector) throws IOException {
DistributedPrimitiveManagerConfiguration managerConfiguration = new DistributedPrimitiveManagerConfiguration(FileBasedPrimitiveManager.class.getName(), Collections.singletonMap("locks-folder", temporaryFolder.newFolder("manager").toString()));
DistributedLockManagerConfiguration managerConfiguration = new DistributedLockManagerConfiguration(FileBasedLockManager.class.getName(), Collections.singletonMap("locks-folder", temporaryFolder.newFolder("manager").toString()));
ReplicatedBackupUtils.configurePluggableQuorumReplicationPair(backupConfig, backupConnector, backupAcceptor,
primaryConfig, primaryConnector, null,

View File

@ -14,11 +14,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.integration.cluster.failover.quorum;
package org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager;
import org.apache.activemq.artemis.tests.integration.cluster.failover.ReplicatedDistributionTest;
public class PluggableQuorumReplicatedDistributionTest extends ReplicatedDistributionTest {
public class LockManagerReplicatedDistributionTest extends ReplicatedDistributionTest {
@Override
protected HAType haType() {

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.integration.cluster.failover.quorum;
package org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager;
import org.apache.activemq.artemis.api.core.client.ClientSession;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
@ -24,9 +24,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.invoke.MethodHandles;
import static org.apache.activemq.artemis.tests.integration.cluster.failover.quorum.PluggableQuorumNettyNoGroupNameReplicatedFailoverTest.doDecrementActivationSequenceForForceRestartOf;
import static org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager.LockManagerNettyNoGroupNameReplicatedFailoverTest.doDecrementActivationSequenceForForceRestartOf;
public class PluggableQuorumReplicatedLargeMessageFailoverTest extends LargeMessageFailoverTest {
public class LockManagerReplicatedLargeMessageFailoverTest extends LargeMessageFailoverTest {
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@Override

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.integration.cluster.failover.quorum;
package org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager;
import org.apache.activemq.artemis.api.core.client.ClientSession;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
@ -22,7 +22,7 @@ import org.apache.activemq.artemis.tests.integration.cluster.util.BackupSyncDela
import org.junit.After;
import org.junit.Before;
public class PluggableQuorumReplicatedLargeMessageWithDelayFailoverTest extends PluggableQuorumReplicatedLargeMessageFailoverTest {
public class LockManagerReplicatedLargeMessageWithDelayFailoverTest extends LockManagerReplicatedLargeMessageFailoverTest {
private BackupSyncDelay syncDelay;

View File

@ -14,12 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.tests.integration.cluster.failover.quorum;
package org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.tests.integration.cluster.failover.PagingFailoverTest;
public class PluggableQuorumReplicatedPagingFailoverTest extends PagingFailoverTest {
public class LockManagerReplicatedPagingFailoverTest extends PagingFailoverTest {
@Override
protected void createConfigs() throws Exception {

View File

@ -20,19 +20,19 @@ import java.io.IOException;
import java.util.Collections;
import org.apache.activemq.artemis.core.config.HAPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedPrimitiveManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationPrimaryPolicyConfiguration;
import org.apache.activemq.artemis.quorum.file.FileBasedPrimitiveManager;
import org.apache.activemq.artemis.lockmanager.file.FileBasedLockManager;
import org.junit.Before;
public class PluggableQuorumReplicationFlowControlTest extends SharedNothingReplicationFlowControlTest {
public class LockManagerReplicationFlowControlTest extends SharedNothingReplicationFlowControlTest {
private DistributedPrimitiveManagerConfiguration managerConfiguration;
private DistributedLockManagerConfiguration managerConfiguration;
@Before
public void init() throws IOException {
managerConfiguration = new DistributedPrimitiveManagerConfiguration(FileBasedPrimitiveManager.class.getName(), Collections.singletonMap("locks-folder", temporaryFolder.newFolder("manager").toString()));
managerConfiguration = new DistributedLockManagerConfiguration(FileBasedLockManager.class.getName(), Collections.singletonMap("locks-folder", temporaryFolder.newFolder("manager").toString()));
}
@Override

View File

@ -24,7 +24,7 @@ import org.junit.runners.Parameterized;
import static java.util.Arrays.asList;
@RunWith(Parameterized.class)
public class PluggableQuorumReplicationOrderTest extends ReplicationOrderTest {
public class LockManagerReplicationOrderTest extends ReplicationOrderTest {
@Parameterized.Parameter
public boolean useNetty;

View File

@ -33,29 +33,29 @@ import org.apache.activemq.artemis.api.core.client.ServerLocator;
import org.apache.activemq.artemis.core.client.impl.ServerLocatorImpl;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.HAPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedPrimitiveManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.DistributedLockManagerConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicationPrimaryPolicyConfiguration;
import org.apache.activemq.artemis.core.server.ActivateCallback;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.ActiveMQServers;
import org.apache.activemq.artemis.core.server.impl.FileLockNodeManager;
import org.apache.activemq.artemis.quorum.DistributedLock;
import org.apache.activemq.artemis.quorum.DistributedPrimitiveManager;
import org.apache.activemq.artemis.quorum.MutableLong;
import org.apache.activemq.artemis.quorum.file.FileBasedPrimitiveManager;
import org.apache.activemq.artemis.lockmanager.DistributedLock;
import org.apache.activemq.artemis.lockmanager.DistributedLockManager;
import org.apache.activemq.artemis.lockmanager.MutableLong;
import org.apache.activemq.artemis.lockmanager.file.FileBasedLockManager;
import org.apache.activemq.artemis.tests.util.Wait;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
public class PluggableQuorumReplicationTest extends SharedNothingReplicationTest {
public class LockManagerReplicationTest extends SharedNothingReplicationTest {
private DistributedPrimitiveManagerConfiguration managerConfiguration;
private DistributedLockManagerConfiguration managerConfiguration;
@Before
public void init() throws IOException {
managerConfiguration = new DistributedPrimitiveManagerConfiguration(FileBasedPrimitiveManager.class.getName(), Collections.singletonMap("locks-folder", temporaryFolder.newFolder("manager").toString()));
managerConfiguration = new DistributedLockManagerConfiguration(FileBasedLockManager.class.getName(), Collections.singletonMap("locks-folder", temporaryFolder.newFolder("manager").toString()));
}
@Override
@ -370,11 +370,11 @@ public class PluggableQuorumReplicationTest extends SharedNothingReplicationTest
// backup can get lock but does not have the sequence to start, will try and be a backup
// to verify it can short circuit with a dirty read we grab the lock for a little while
DistributedPrimitiveManager distributedPrimitiveManager = DistributedPrimitiveManager.newInstanceOf(
DistributedLockManager distributedLockManager = DistributedLockManager.newInstanceOf(
managerConfiguration.getClassName(),
managerConfiguration.getProperties());
distributedPrimitiveManager.start();
final DistributedLock lock = distributedPrimitiveManager.getDistributedLock(coordinatedId);
distributedLockManager.start();
final DistributedLock lock = distributedLockManager.getDistributedLock(coordinatedId);
assertTrue(lock.tryLock());
CountDownLatch preActivate = new CountDownLatch(1);
backupServer.registerActivateCallback(new ActivateCallback() {
@ -390,7 +390,7 @@ public class PluggableQuorumReplicationTest extends SharedNothingReplicationTest
assertTrue(preActivate.await(1, TimeUnit.SECONDS));
// release the lock
distributedPrimitiveManager.stop();
distributedLockManager.stop();
// primary server should be active
primaryServer.start();
@ -419,19 +419,18 @@ public class PluggableQuorumReplicationTest extends SharedNothingReplicationTest
// backup can get lock but does not have the sequence to start, will try and be a backup
// to verify it can short circuit with a dirty read we grab the lock for a little while
DistributedPrimitiveManager distributedPrimitiveManager = DistributedPrimitiveManager
.newInstanceOf(managerConfiguration.getClassName(), managerConfiguration.getProperties());
distributedPrimitiveManager.start();
try (DistributedLock lock = distributedPrimitiveManager.getDistributedLock(coordinatedId)) {
DistributedLockManager distributedLockManager = DistributedLockManager.newInstanceOf(managerConfiguration.getClassName(), managerConfiguration.getProperties());
distributedLockManager.start();
try (DistributedLock lock = distributedLockManager.getDistributedLock(coordinatedId)) {
assertTrue(lock.tryLock());
distributedPrimitiveManager.getMutableLong(coordinatedId).compareAndSet(2, -2);
distributedLockManager.getMutableLong(coordinatedId).compareAndSet(2, -2);
}
primaryServer.start();
Wait.waitFor(primaryServer::isStarted);
Assert.assertEquals(3, primaryServer.getNodeManager().getNodeActivationSequence());
Assert.assertEquals(3, distributedPrimitiveManager.getMutableLong(coordinatedId).get());
Assert.assertEquals(3, distributedLockManager.getMutableLong(coordinatedId).get());
distributedPrimitiveManager.stop();
distributedLockManager.stop();
Configuration backupConfiguration = createBackupConfiguration();
ActiveMQServer backupServer = addServer(ActiveMQServers.newActiveMQServer(backupConfiguration));
@ -570,9 +569,9 @@ public class PluggableQuorumReplicationTest extends SharedNothingReplicationTest
// some manual intervention to force an unavailable
// simulate primary failing in activation local sequence update on un replicated run when backup stops.
DistributedPrimitiveManager distributedPrimitiveManager = DistributedPrimitiveManager.newInstanceOf(managerConfiguration.getClassName(), managerConfiguration.getProperties());
distributedPrimitiveManager.start();
final MutableLong activationSequence = distributedPrimitiveManager.getMutableLong(coordinatedId);
DistributedLockManager distributedLockManager = DistributedLockManager.newInstanceOf(managerConfiguration.getClassName(), managerConfiguration.getProperties());
distributedLockManager.start();
final MutableLong activationSequence = distributedLockManager.getMutableLong(coordinatedId);
Assert.assertTrue(activationSequence.compareAndSet(2, -2));
// primary server should activate after self healing its outstanding claim
@ -612,11 +611,11 @@ public class PluggableQuorumReplicationTest extends SharedNothingReplicationTest
// some manual intervention to force an unavailable
// simulate primary failing in activation local sequence update on un replicated run when backup stops.
DistributedPrimitiveManager distributedPrimitiveManager = DistributedPrimitiveManager.newInstanceOf(
DistributedLockManager distributedLockManager = DistributedLockManager.newInstanceOf(
managerConfiguration.getClassName(),
managerConfiguration.getProperties());
distributedPrimitiveManager.start();
final MutableLong coordinatedActivationSequence = distributedPrimitiveManager.getMutableLong(coordinatedId);
distributedLockManager.start();
final MutableLong coordinatedActivationSequence = distributedLockManager.getMutableLong(coordinatedId);
Assert.assertTrue(coordinatedActivationSequence.compareAndSet(2, -2));
// case: 2, the fail to write locally 2 but the write actually failing

Some files were not shown because too many files have changed in this diff Show More