diff --git a/core/src/main/java/org/infinispan/configuration/global/GlobalConfiguration.java b/core/src/main/java/org/infinispan/configuration/global/GlobalConfiguration.java
index bc8c402f5587..7c4f0b89a8d0 100644
--- a/core/src/main/java/org/infinispan/configuration/global/GlobalConfiguration.java
+++ b/core/src/main/java/org/infinispan/configuration/global/GlobalConfiguration.java
@@ -31,6 +31,8 @@
 @Scope(Scopes.GLOBAL)
 @SurvivesRestarts
 public class GlobalConfiguration {
+   private static final String ZERO_CAPACITY_NODE_FEATURE = "zero-capacity-node";
+
 
    /**
     * Default replication version, from {@link org.infinispan.Version#getVersionShort}.
@@ -57,6 +59,7 @@ public class GlobalConfiguration {
    private final ThreadPoolConfiguration asyncThreadPool;
    private final Optional<String> defaultCacheName;
    private final Features features;
+   private final boolean zeroCapacityNode;
 
    GlobalConfiguration(ThreadPoolConfiguration expirationThreadPool,
                        ThreadPoolConfiguration listenerThreadPool,
@@ -68,7 +71,10 @@ public class GlobalConfiguration {
                        TransportConfiguration transport, GlobalSecurityConfiguration security,
                        SerializationConfiguration serialization, ShutdownConfiguration shutdown,
                        GlobalStateConfiguration globalState,
-                       List<?> modules, SiteConfiguration site, Optional<String> defaultCacheName, ClassLoader cl, Features features) {
+                       List<?> modules, SiteConfiguration site,
+                       Optional<String> defaultCacheName,
+                       ClassLoader cl, Features features,
+                       boolean zeroCapacityNode) {
       this.expirationThreadPool = expirationThreadPool;
       this.listenerThreadPool = listenerThreadPool;
       this.replicationQueueThreadPool = replicationQueueThreadPool;
@@ -82,7 +88,7 @@ public class GlobalConfiguration {
       this.shutdown = shutdown;
       this.globalState = globalState;
       Map<Class<?>, Object> moduleMap = new HashMap<>();
-      for(Object module : modules) {
+      for (Object module : modules) {
          moduleMap.put(module.getClass(), module);
       }
       this.modules = Collections.unmodifiableMap(moduleMap);
@@ -90,6 +96,7 @@ public class GlobalConfiguration {
       this.defaultCacheName = defaultCacheName;
       this.cl = cl;
       this.features = features;
+      this.zeroCapacityNode = features.isAvailable(ZERO_CAPACITY_NODE_FEATURE) ? zeroCapacityNode : false;
    }
 
    /**
@@ -201,7 +208,7 @@ public GlobalStateConfiguration globalState() {
 
    @SuppressWarnings("unchecked")
    public <T> T module(Class<T> moduleClass) {
-      return (T)modules.get(moduleClass);
+      return (T) modules.get(moduleClass);
    }
 
    public Map<Class<?>, ?> modules() {
@@ -245,10 +252,15 @@ public String toString() {
             ", site=" + site +
             ", defaultCacheName=" + defaultCacheName +
             ", cl=" + cl +
+            ", zeroCapacityNode=" + zeroCapacityNode +
             '}';
    }
 
    public boolean isClustered() {
       return transport().transport() != null;
    }
+
+   public boolean isZeroCapacityNode() {
+      return zeroCapacityNode;
+   }
 }
diff --git a/core/src/main/java/org/infinispan/configuration/global/GlobalConfigurationBuilder.java b/core/src/main/java/org/infinispan/configuration/global/GlobalConfigurationBuilder.java
index bb09cb9770e1..b5b2e1d363b9 100644
--- a/core/src/main/java/org/infinispan/configuration/global/GlobalConfigurationBuilder.java
+++ b/core/src/main/java/org/infinispan/configuration/global/GlobalConfigurationBuilder.java
@@ -34,6 +34,7 @@ public class GlobalConfigurationBuilder implements GlobalConfigurationChildBuild
    private final Map<Class<?>, Builder<?>> modules;
    private final SiteConfigurationBuilder site;
    private Optional<String> defaultCacheName;
+   private boolean zeroCapacityNode;
    private Features features;
 
    public GlobalConfigurationBuilder() {
@@ -57,6 +58,7 @@ public GlobalConfigurationBuilder() {
       this.asyncThreadPool = new ThreadPoolConfigurationBuilder(this);
       this.modules = new LinkedHashMap();
       this.defaultCacheName = Optional.empty();
+      this.zeroCapacityNode = false;
    }
 
    /**
@@ -165,6 +167,11 @@ public <T> T module(Class<T> moduleClass) {
       return (T)modules.get(moduleClass);
    }
 
+   public GlobalConfigurationBuilder zeroCapacityNode(boolean zeroCapacityNode) {
+      this.zeroCapacityNode = zeroCapacityNode;
+      return this;
+   }
+
    public GlobalConfigurationBuilder clearModules() {
       modules.clear();
       return this;
@@ -259,7 +266,8 @@ public GlobalConfiguration build() {
             site.create(),
             defaultCacheName,
             cl,
-            features);
+            features,
+            zeroCapacityNode);
    }
 
    public GlobalConfigurationBuilder read(GlobalConfiguration template) {
diff --git a/core/src/main/java/org/infinispan/configuration/parsing/Attribute.java b/core/src/main/java/org/infinispan/configuration/parsing/Attribute.java
index 5ff5f57c37fd..ad5de058d8e3 100644
--- a/core/src/main/java/org/infinispan/configuration/parsing/Attribute.java
+++ b/core/src/main/java/org/infinispan/configuration/parsing/Attribute.java
@@ -167,6 +167,7 @@ public enum Attribute {
     WAIT_TIME,
     WHEN_SPLIT,
     WRITE_SKEW_CHECK("write-skew"),
+    ZERO_CAPACITY_NODE("zero-capacity-node")
     ;
 
     private final String name;
diff --git a/core/src/main/java/org/infinispan/configuration/parsing/Parser.java b/core/src/main/java/org/infinispan/configuration/parsing/Parser.java
index 16ede1f4d26e..4cf40eb23271 100644
--- a/core/src/main/java/org/infinispan/configuration/parsing/Parser.java
+++ b/core/src/main/java/org/infinispan/configuration/parsing/Parser.java
@@ -628,13 +628,17 @@ private void parseContainer(XMLExtendedStreamReader reader, ConfigurationBuilder
                break;
             }
             case STATISTICS: {
-               builder.globalJmxStatistics().enabled(Boolean.valueOf(value));
+               builder.globalJmxStatistics().enabled(Boolean.parseBoolean(value));
                break;
             }
             case SHUTDOWN_HOOK: {
                builder.shutdown().hookBehavior(ShutdownHookBehavior.valueOf(value));
                break;
             }
+            case ZERO_CAPACITY_NODE: {
+               builder.zeroCapacityNode(Boolean.parseBoolean(value));
+               break;
+            }
             default: {
                throw ParseUtils.unexpectedAttribute(reader, i);
             }
diff --git a/core/src/main/java/org/infinispan/statetransfer/StateTransferManagerImpl.java b/core/src/main/java/org/infinispan/statetransfer/StateTransferManagerImpl.java
index 00cd3a2debbd..e77e051cb926 100644
--- a/core/src/main/java/org/infinispan/statetransfer/StateTransferManagerImpl.java
+++ b/core/src/main/java/org/infinispan/statetransfer/StateTransferManagerImpl.java
@@ -94,6 +94,8 @@ public void start() throws Exception {
          persistentStateChecksum = Optional.empty();
       }
 
+      float capacityFactor = globalConfiguration.isZeroCapacityNode() ? 0.0f : configuration.clustering().hash().capacityFactor();
+
       CacheJoinInfo joinInfo = new CacheJoinInfo(pickConsistentHashFactory(globalConfiguration, configuration),
             configuration.clustering().hash().hash(),
             configuration.clustering().hash().numSegments(),
@@ -101,7 +103,7 @@ public void start() throws Exception {
             configuration.clustering().stateTransfer().timeout(),
             configuration.transaction().transactionProtocol().isTotalOrder(),
             configuration.clustering().cacheMode(),
-            configuration.clustering().hash().capacityFactor(),
+            capacityFactor,
             localTopologyManager.getPersistentUUID(),
             persistentStateChecksum);
 
diff --git a/core/src/main/resources/schema/infinispan-config-10.0.xsd b/core/src/main/resources/schema/infinispan-config-10.0.xsd
index 92d5ab16ee2f..e458d9e90702 100644
--- a/core/src/main/resources/schema/infinispan-config-10.0.xsd
+++ b/core/src/main/resources/schema/infinispan-config-10.0.xsd
@@ -315,6 +315,11 @@
         <xs:documentation>Indicates the default cache for this cache container</xs:documentation>
       </xs:annotation>
     </xs:attribute>
+    <xs:attribute name="zero-capacity-node" type="xs:boolean" default="false">
+      <xs:annotation>
+        <xs:documentation>If 'true' then no data is stored in this node. Defaults to 'false'.</xs:documentation>
+      </xs:annotation>
+    </xs:attribute>
     <xs:attribute name="start">
       <xs:annotation><xs:documentation>Unused XML attribute</xs:documentation></xs:annotation>
     </xs:attribute>
diff --git a/core/src/test/java/org/infinispan/distribution/ZeroCapacityNodeTest.java b/core/src/test/java/org/infinispan/distribution/ZeroCapacityNodeTest.java
new file mode 100644
index 000000000000..0089baeea9d6
--- /dev/null
+++ b/core/src/test/java/org/infinispan/distribution/ZeroCapacityNodeTest.java
@@ -0,0 +1,59 @@
+package org.infinispan.distribution;
+
+import static org.testng.AssertJUnit.assertEquals;
+
+import java.util.Map;
+
+import org.infinispan.configuration.cache.CacheMode;
+import org.infinispan.configuration.cache.ConfigurationBuilder;
+import org.infinispan.configuration.global.GlobalConfigurationBuilder;
+import org.infinispan.distribution.ch.ConsistentHash;
+import org.infinispan.distribution.ch.impl.DefaultConsistentHash;
+import org.infinispan.distribution.group.impl.PartitionerConsistentHash;
+import org.infinispan.manager.EmbeddedCacheManager;
+import org.infinispan.remoting.transport.Address;
+import org.infinispan.test.MultipleCacheManagersTest;
+import org.infinispan.test.TestingUtil;
+import org.testng.annotations.Test;
+
+/**
+ * Test the capacity factor for lite instance
+ *
+ * @author Katia Aresti
+ * @since 9.4
+ */
+@Test(groups = "functional", testName = "distribution.ch.ZeroCapacityNodeTest")
+public class ZeroCapacityNodeTest extends MultipleCacheManagersTest {
+
+   public static final int NUM_SEGMENTS = 60;
+
+   @Override
+   protected void createCacheManagers() throws Throwable {
+      // Do nothing here, create the cache managers in the test
+   }
+
+   public void testCapacityFactorContainingAZeroCapacityNode() {
+
+      ConfigurationBuilder cb = new ConfigurationBuilder();
+      cb.clustering().cacheMode(CacheMode.DIST_SYNC);
+      cb.clustering().hash().numSegments(NUM_SEGMENTS);
+      cb.clustering().hash().capacityFactor(0.5f);
+
+      EmbeddedCacheManager node1 = addClusterEnabledCacheManager(GlobalConfigurationBuilder.defaultClusteredBuilder(), cb);
+      EmbeddedCacheManager node2 = addClusterEnabledCacheManager(GlobalConfigurationBuilder.defaultClusteredBuilder(), cb);
+      EmbeddedCacheManager zeroCapacityNode = addClusterEnabledCacheManager(GlobalConfigurationBuilder.defaultClusteredBuilder().zeroCapacityNode(true), cb);
+
+      waitForClusterToForm();
+      assertCapacityFactors(node1, 0.5f);
+      assertCapacityFactors(node2, 0.5f);
+      assertCapacityFactors(zeroCapacityNode, 0.0f);
+   }
+
+   private void assertCapacityFactors(EmbeddedCacheManager cm, float expectedCapacityFactors) {
+      ConsistentHash ch = cache(0).getAdvancedCache().getDistributionManager().getReadConsistentHash();
+      DefaultConsistentHash dch =
+            (DefaultConsistentHash) TestingUtil.extractField(PartitionerConsistentHash.class, ch, "ch");
+      Map<Address, Float> capacityFactors = dch.getCapacityFactors();
+      assertEquals(expectedCapacityFactors, capacityFactors.get(cm.getAddress()), 0.0);
+   }
+}
diff --git a/core/src/test/java/org/infinispan/functional/FunctionalTestUtils.java b/core/src/test/java/org/infinispan/functional/FunctionalTestUtils.java
index 3d33df4261fe..6c9cc03a284c 100644
--- a/core/src/test/java/org/infinispan/functional/FunctionalTestUtils.java
+++ b/core/src/test/java/org/infinispan/functional/FunctionalTestUtils.java
@@ -5,6 +5,7 @@
 import static org.testng.AssertJUnit.assertTrue;
 import static org.testng.AssertJUnit.fail;
 
+import java.util.List;
 import java.util.Random;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -18,6 +19,7 @@
 import org.infinispan.functional.impl.ReadOnlyMapImpl;
 import org.infinispan.functional.impl.ReadWriteMapImpl;
 import org.infinispan.functional.impl.WriteOnlyMapImpl;
+import org.infinispan.util.concurrent.CompletableFutures;
 
 public final class FunctionalTestUtils {
 
@@ -48,6 +50,10 @@ public static <T> T await(CompletableFuture<T> cf) {
       }
    }
 
+   public static <T> List<T> await(List<CompletableFuture<T>> cf) {
+      return await(CompletableFutures.sequence(cf));
+   }
+
    public static <K> void assertReadOnlyViewEmpty(K k, ReadEntryView<K, ?> ro) {
       assertEquals(k, ro.key());
       assertFalse(ro.find().isPresent());
diff --git a/core/src/test/resources/configs/unified/10.0.xml b/core/src/test/resources/configs/unified/10.0.xml
index 9284e3b3e2a1..6d58e96b2de2 100644
--- a/core/src/test/resources/configs/unified/10.0.xml
+++ b/core/src/test/resources/configs/unified/10.0.xml
@@ -97,7 +97,7 @@
 
    <cache-container name="maximal" aliases="alias1 alias2" default-cache="local" async-executor="infinispan-async" expiration-executor="infinispan-expiration"
                     jndi-name="java:global/infinispan/maximal"  state-transfer-executor="infinispan-state-transfer" listener-executor="infinispan-listener"
-                    persistence-executor="infinispan-cached" module="org.infinispan" statistics="true" shutdown-hook="DONT_REGISTER">
+                    persistence-executor="infinispan-cached" module="org.infinispan" statistics="true" shutdown-hook="DONT_REGISTER" zero-capacity-node="false">
       <transport cluster="maximal-cluster" executor="infinispan-transport" remote-command-executor="infinispan-cached" lock-timeout="120000" stack="tcp" node-name="a-node" machine="a" rack="b" site="c"
                  initial-cluster-size="4" initial-cluster-timeout="30000" />
       <serialization marshaller="org.infinispan.marshall.TestObjectStreamMarshaller" version="1.0">
diff --git a/core/src/test/resources/infinispan-features.properties b/core/src/test/resources/infinispan-features.properties
new file mode 100644
index 000000000000..796e8924ea22
--- /dev/null
+++ b/core/src/test/resources/infinispan-features.properties
@@ -0,0 +1 @@
+org.infinispan.feature.zero-capacity-node=true
\ No newline at end of file
diff --git a/counter/src/test/java/org/infinispan/counter/StrongCounterWithZeroCapacityNodesTest.java b/counter/src/test/java/org/infinispan/counter/StrongCounterWithZeroCapacityNodesTest.java
new file mode 100644
index 000000000000..c8df4d8849de
--- /dev/null
+++ b/counter/src/test/java/org/infinispan/counter/StrongCounterWithZeroCapacityNodesTest.java
@@ -0,0 +1,20 @@
+package org.infinispan.counter;
+
+import org.infinispan.configuration.global.GlobalConfigurationBuilder;
+import org.testng.annotations.Test;
+
+/**
+ * A simple consistency test for {@link org.infinispan.counter.api.StrongCounter} where some nodes are capacity factor
+ * 0.
+ *
+ * @author Katia Aresti, karesti@redhat.com
+ * @since 9.4
+ */
+@Test(groups = "functional", testName = "counter.StrongCounterWithZeroCapacityNodesTest")
+public class StrongCounterWithZeroCapacityNodesTest extends StrongCounterTest {
+
+   @Override
+   protected GlobalConfigurationBuilder configure(int nodeId) {
+      return GlobalConfigurationBuilder.defaultClusteredBuilder().zeroCapacityNode(nodeId % 2 == 0);
+   }
+}
diff --git a/counter/src/test/java/org/infinispan/counter/WeakCounterWithZeroCapacityNodesTest.java b/counter/src/test/java/org/infinispan/counter/WeakCounterWithZeroCapacityNodesTest.java
new file mode 100644
index 000000000000..c2bee5329fde
--- /dev/null
+++ b/counter/src/test/java/org/infinispan/counter/WeakCounterWithZeroCapacityNodesTest.java
@@ -0,0 +1,20 @@
+package org.infinispan.counter;
+
+import org.infinispan.configuration.global.GlobalConfigurationBuilder;
+import org.infinispan.counter.api.WeakCounter;
+import org.testng.annotations.Test;
+
+/**
+ * A simple consistency test for {@link WeakCounter} with zero capacity nodes.
+ *
+ * @author Katia Aresti, karesti@redhat.com
+ * @since 9.4
+ */
+@Test(groups = "functional", testName = "counter.WeakCounterWithZeroCapacityNodesTest")
+public class WeakCounterWithZeroCapacityNodesTest extends WeakCounterTest {
+
+   @Override
+   protected GlobalConfigurationBuilder configure(int nodeId) {
+      return GlobalConfigurationBuilder.defaultClusteredBuilder().zeroCapacityNode(nodeId % 2 == 0);
+   }
+}
diff --git a/documentation/src/main/asciidoc/user_guide/clustering.adoc b/documentation/src/main/asciidoc/user_guide/clustering.adoc
index 1f00e24c5868..983ac84d473b 100644
--- a/documentation/src/main/asciidoc/user_guide/clustering.adoc
+++ b/documentation/src/main/asciidoc/user_guide/clustering.adoc
@@ -342,6 +342,21 @@ With cross-site replication as well, the "site master" should only deal with for
 commands between sites and shouldn't handle user requests, so it makes sense to configure
 it with a capacity factor of `0`.
 
+===== Zero Capacity Node
+You might need to configure a whole node where the capacity factor is `0` for every cache,
+user defined caches and internal caches.
+When defining a zero capacity node, the node won't hold any data.
+This is how you declare a zero capacity node:
+
+[source,xml]
+----
+   <cache-container zero-capacity-node="true" />
+----
+
+[source, java]
+----
+new GlobalConfigurationBuilder().zeroCapacityNode(true);
+----
 
 ===== Hashing Configuration
 This is how you configure hashing declaratively, via XML:
diff --git a/lock/src/main/java/org/infinispan/lock/impl/ClusteredLockModuleLifecycle.java b/lock/src/main/java/org/infinispan/lock/impl/ClusteredLockModuleLifecycle.java
index be8da02691d2..d3ca1ca426c4 100644
--- a/lock/src/main/java/org/infinispan/lock/impl/ClusteredLockModuleLifecycle.java
+++ b/lock/src/main/java/org/infinispan/lock/impl/ClusteredLockModuleLifecycle.java
@@ -5,6 +5,7 @@
 import java.util.concurrent.CompletableFuture;
 
 import org.infinispan.Cache;
+import org.infinispan.commons.logging.LogFactory;
 import org.infinispan.commons.marshall.AdvancedExternalizer;
 import org.infinispan.configuration.cache.CacheMode;
 import org.infinispan.configuration.cache.Configuration;
@@ -25,6 +26,7 @@
 import org.infinispan.lock.impl.lock.ClusteredLockFilter;
 import org.infinispan.lock.impl.manager.CacheHolder;
 import org.infinispan.lock.impl.manager.EmbeddedClusteredLockManager;
+import org.infinispan.lock.logging.Log;
 import org.infinispan.manager.EmbeddedCacheManager;
 import org.infinispan.partitionhandling.PartitionHandling;
 import org.infinispan.registry.InternalCacheRegistry;
@@ -39,6 +41,7 @@
  */
 @MetaInfServices(value = ModuleLifecycle.class)
 public class ClusteredLockModuleLifecycle implements ModuleLifecycle {
+   private static final Log log = LogFactory.getLog(ClusteredLockModuleLifecycle.class, Log.class);
 
    public static final String CLUSTERED_LOCK_CACHE_NAME = "org.infinispan.LOCKS";
 
@@ -61,8 +64,9 @@ public void cacheManagerStarted(GlobalComponentRegistry gcr) {
       final InternalCacheRegistry internalCacheRegistry = gcr.getComponent(InternalCacheRegistry.class);
 
       ClusteredLockManagerConfiguration config = extractConfiguration(gcr);
+      GlobalConfiguration globalConfig = gcr.getGlobalConfiguration();
 
-      internalCacheRegistry.registerInternalCache(CLUSTERED_LOCK_CACHE_NAME, createClusteredLockCacheConfiguration(config),
+      internalCacheRegistry.registerInternalCache(CLUSTERED_LOCK_CACHE_NAME, createClusteredLockCacheConfiguration(config, globalConfig),
             EnumSet.of(InternalCacheRegistry.Flag.EXCLUSIVE));
 
       CompletableFuture<CacheHolder> future = startCaches(cacheManager);
@@ -75,13 +79,15 @@ private static ClusteredLockManagerConfiguration extractConfiguration(GlobalComp
       return config == null ? ClusteredLockManagerConfigurationBuilder.defaultConfiguration() : config;
    }
 
-   private static Configuration createClusteredLockCacheConfiguration(ClusteredLockManagerConfiguration config) {
+   private static Configuration createClusteredLockCacheConfiguration(ClusteredLockManagerConfiguration config, GlobalConfiguration globalConfig) {
       ConfigurationBuilder builder = new ConfigurationBuilder();
       builder.transaction().transactionMode(TransactionMode.NON_TRANSACTIONAL);
 
       if (config.numOwners() > 0) {
          builder.clustering().cacheMode(CacheMode.DIST_SYNC)
                .hash().numOwners(config.numOwners());
+      } else if (globalConfig.isZeroCapacityNode()) {
+         throw log.zeroCapacityNodeError();
       } else {
          builder.clustering().cacheMode(CacheMode.REPL_SYNC);
       }
diff --git a/lock/src/main/java/org/infinispan/lock/logging/Log.java b/lock/src/main/java/org/infinispan/lock/logging/Log.java
index 305da879e24a..c2f63235cd66 100644
--- a/lock/src/main/java/org/infinispan/lock/logging/Log.java
+++ b/lock/src/main/java/org/infinispan/lock/logging/Log.java
@@ -37,4 +37,7 @@ public interface Log extends BasicLogger {
 
    @Message(value = "Invalid scope for tag <clustered-lock>. Expected CACHE_CONTAINER but was %s", id = 29007)
    ClusteredLockException invalidScope(String scope);
+
+   @Message(value = "When the node is configured as a zero-capacity node, you need to specify the number of owners for the lock", id = 29008)
+   ClusteredLockException zeroCapacityNodeError();
 }
diff --git a/lock/src/test/java/org/infinispan/lock/ClusteredLockTest.java b/lock/src/test/java/org/infinispan/lock/ClusteredLockTest.java
index 389d39570fb1..50a42314f6d5 100644
--- a/lock/src/test/java/org/infinispan/lock/ClusteredLockTest.java
+++ b/lock/src/test/java/org/infinispan/lock/ClusteredLockTest.java
@@ -9,8 +9,10 @@
 import static org.testng.AssertJUnit.fail;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
 
 import org.infinispan.configuration.cache.CacheMode;
 import org.infinispan.lock.api.ClusteredLock;
@@ -211,66 +213,29 @@ public void testTryLockWithTimeoutWithCountersInParallelOnMultiLocks() throws Th
 
    @Test
    public void testTryLockWithCountersInParallel() throws Throwable {
-      AtomicInteger counter = new AtomicInteger();
-
       ClusteredLock lock0 = clusteredLockManager(0).get(LOCK_NAME);
       ClusteredLock lock1 = clusteredLockManager(1).get(LOCK_NAME);
       ClusteredLock lock2 = clusteredLockManager(2).get(LOCK_NAME);
 
-      CompletableFuture<Void> lockRes0 = lock0.tryLock()
-            .thenCompose(result -> {
-               if (result) {
-                  new Counter(counter, 1, 100).run();
-                  return lock0.unlock();
-               }
-               return CompletableFuture.completedFuture(null);
-            });
-
-      CompletableFuture<Void> lockRes1 = lock1.tryLock()
-            .thenCompose(result -> {
-               if (result) {
-                  new Counter(counter, 1, 100).run();
-                  return lock1.unlock();
-               }
-               return CompletableFuture.completedFuture(null);
-            });
-
-      CompletableFuture<Void> lockRes2 = lock2.tryLock()
-            .thenCompose(result -> {
-               if (result) {
-                  new Counter(counter, 1, 100).run();
-                  return lock2.unlock();
-               }
-               return CompletableFuture.completedFuture(null);
-            });
-
-      await(lockRes0);
-      await(lockRes1);
-      await(lockRes2);
+      long successTryLocks = Stream.of(lock0, lock1, lock2)
+            .map(this::callTryLock)
+            .map(ClusteredLockTest::awaitTryLock)
+            .filter(Boolean::booleanValue)
+            .count();
 
-      assertEquals(1, counter.get());
+      assertEquals(1, successTryLocks);
    }
 
-   class Counter implements Runnable {
-      private AtomicInteger counter;
-      private int delta;
-      private long millis;
-
-      Counter(AtomicInteger counter, int delta, long millis) {
-         this.counter = counter;
-         this.delta = delta;
-         this.millis = millis;
-      }
+   private Future<Boolean> callTryLock(ClusteredLock lock) {
+      return fork(() -> await(lock.tryLock()));
+   }
 
-      @Override
-      public void run() {
-         // Sleep a while
-         try {
-            TimeUnit.MILLISECONDS.sleep(millis);
-         } catch (InterruptedException e) {
-            fail("There was a problem in the Counter");
-         }
-         counter.addAndGet(delta);
+   private static Boolean awaitTryLock(Future<Boolean> result) {
+      try {
+         return result.get();
+      } catch (Exception e) {
+         fail("tryLock call should work and return either TRUE ou FALSE");
       }
+      return null;
    }
 }
diff --git a/lock/src/test/java/org/infinispan/lock/ClusteredLockWithZeroCapacityIncorrectConfigTest.java b/lock/src/test/java/org/infinispan/lock/ClusteredLockWithZeroCapacityIncorrectConfigTest.java
new file mode 100644
index 000000000000..2b7d47eebdbf
--- /dev/null
+++ b/lock/src/test/java/org/infinispan/lock/ClusteredLockWithZeroCapacityIncorrectConfigTest.java
@@ -0,0 +1,49 @@
+package org.infinispan.lock;
+
+import org.infinispan.configuration.cache.CacheMode;
+import org.infinispan.configuration.global.GlobalConfigurationBuilder;
+import org.infinispan.lock.exception.ClusteredLockException;
+import org.infinispan.manager.EmbeddedCacheManagerStartupException;
+import org.infinispan.test.Exceptions;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Test(groups = "functional", testName = "clusteredLock.ClusteredLockWithZeroCapacityIncorrectConfigTest")
+public class ClusteredLockWithZeroCapacityIncorrectConfigTest extends BaseClusteredLockTest {
+
+   public ClusteredLockWithZeroCapacityIncorrectConfigTest() {
+      super();
+      numOwner = -1;
+      cacheMode = CacheMode.DIST_SYNC;
+   }
+
+   @Override
+   protected int clusterSize() {
+      return 3;
+   }
+
+   @BeforeClass(alwaysRun = true)
+   @Override
+   public void createBeforeClass() throws Throwable {
+      // Nothing
+   }
+
+   @Override
+   protected GlobalConfigurationBuilder configure(int nodeId) {
+      return super.configure(nodeId).zeroCapacityNode(nodeId == 1);
+   }
+
+   public void testClusterStartupError() {
+      Exceptions.expectException(RuntimeException.class,
+            EmbeddedCacheManagerStartupException.class,
+            ClusteredLockException.class, this::createCluster);
+   }
+
+   private void createCluster() {
+      try {
+         createCacheManagers();
+      } catch (Throwable throwable) {
+         throw new RuntimeException(throwable);
+      }
+   }
+}
diff --git a/lock/src/test/java/org/infinispan/lock/ClusteredLockWithZeroCapacityNodesTest.java b/lock/src/test/java/org/infinispan/lock/ClusteredLockWithZeroCapacityNodesTest.java
new file mode 100644
index 000000000000..9da6de813f5e
--- /dev/null
+++ b/lock/src/test/java/org/infinispan/lock/ClusteredLockWithZeroCapacityNodesTest.java
@@ -0,0 +1,23 @@
+package org.infinispan.lock;
+
+import org.infinispan.configuration.cache.CacheMode;
+import org.infinispan.configuration.global.GlobalConfigurationBuilder;
+import org.testng.annotations.Test;
+
+@Test(groups = "functional", testName = "clusteredLock.ClusteredLockWithZeroCapacityNodesTest")
+public class ClusteredLockWithZeroCapacityNodesTest extends ClusteredLockTest {
+
+   public ClusteredLockWithZeroCapacityNodesTest() {
+      numOwner = 1;
+      cacheMode = CacheMode.DIST_SYNC;
+   }
+
+   protected int clusterSize() {
+      return 3;
+   }
+
+   @Override
+   protected GlobalConfigurationBuilder configure(int nodeId) {
+      return super.configure(nodeId).zeroCapacityNode(nodeId % 2 == 1);
+   }
+}
diff --git a/server/core/src/main/java/org/infinispan/server/core/configuration/ProtocolServerConfiguration.java b/server/core/src/main/java/org/infinispan/server/core/configuration/ProtocolServerConfiguration.java
index c7c31003cd9b..ed15fd8cc08a 100644
--- a/server/core/src/main/java/org/infinispan/server/core/configuration/ProtocolServerConfiguration.java
+++ b/server/core/src/main/java/org/infinispan/server/core/configuration/ProtocolServerConfiguration.java
@@ -31,10 +31,11 @@ public abstract class ProtocolServerConfiguration {
    public static final AttributeDefinition<Integer> IO_THREADS = AttributeDefinition.builder("io-threads", 2 * ProcessorInfo.availableProcessors()).immutable().build();
    public static final AttributeDefinition<Integer> WORKER_THREADS = AttributeDefinition.builder("worker-threads", 160).immutable().build();
    public static final AttributeDefinition<AdminOperationsHandler> ADMIN_OPERATION_HANDLER = AttributeDefinition.builder("admin-operation-handler", null, AdminOperationsHandler.class).immutable().build();
+   public static final AttributeDefinition<Boolean> ZERO_CAPACITY_NODE = AttributeDefinition.builder("zero-capacity-node", false).immutable().build();
 
    public static AttributeSet attributeDefinitionSet() {
       return new AttributeSet(ProtocolServerConfiguration.class,
-            DEFAULT_CACHE_NAME, NAME, HOST, PORT, IDLE_TIMEOUT, IGNORED_CACHES, RECV_BUF_SIZE, SEND_BUF_SIZE, START_TRANSPORT, TCP_NODELAY, TCP_KEEPALIVE, IO_THREADS, WORKER_THREADS, ADMIN_OPERATION_HANDLER);
+            DEFAULT_CACHE_NAME, NAME, HOST, PORT, IDLE_TIMEOUT, IGNORED_CACHES, RECV_BUF_SIZE, SEND_BUF_SIZE, START_TRANSPORT, TCP_NODELAY, TCP_KEEPALIVE, IO_THREADS, WORKER_THREADS, ADMIN_OPERATION_HANDLER, ZERO_CAPACITY_NODE);
    }
 
    private final Attribute<String> defaultCacheName;
@@ -51,6 +52,7 @@ public static AttributeSet attributeDefinitionSet() {
    private final Attribute<Set<String>> ignoredCaches;
    private final Attribute<Boolean> startTransport;
    private final Attribute<AdminOperationsHandler> adminOperationsHandler;
+   private final Attribute<Boolean> zeroCapacityNode;
 
    protected final AttributeSet attributes;
 
@@ -61,6 +63,7 @@ protected ProtocolServerConfiguration(AttributeSet attributes, SslConfiguration
       this.ssl = ssl;
 
       defaultCacheName = attributes.attribute(DEFAULT_CACHE_NAME);
+      zeroCapacityNode = attributes.attribute(ZERO_CAPACITY_NODE);
       name = attributes.attribute(NAME);
       host = attributes.attribute(HOST);
       port = attributes.attribute(PORT);
@@ -140,6 +143,10 @@ public AdminOperationsHandler adminOperationsHandler() {
       return adminOperationsHandler.get();
    }
 
+   public boolean zeroCapacityNode() {
+      return zeroCapacityNode.get();
+   }
+
    @Override
    public String toString() {
       return "ProtocolServerConfiguration[" + attributes + "]";