diff --git a/hazelcast/src/main/java/com/hazelcast/internal/partition/membergroup/SPIAwareMemberGroupFactory.java b/hazelcast/src/main/java/com/hazelcast/internal/partition/membergroup/SPIAwareMemberGroupFactory.java
index 1dd8200d59cd..0de9fdcb531c 100644
--- a/hazelcast/src/main/java/com/hazelcast/internal/partition/membergroup/SPIAwareMemberGroupFactory.java
+++ b/hazelcast/src/main/java/com/hazelcast/internal/partition/membergroup/SPIAwareMemberGroupFactory.java
@@ -25,6 +25,7 @@
 import com.hazelcast.spi.discovery.impl.DefaultDiscoveryService;
 import com.hazelcast.spi.discovery.integration.DiscoveryService;
 import com.hazelcast.spi.partitiongroup.MemberGroup;
+import com.hazelcast.spi.partitiongroup.PartitionGroupStrategy;
 
 import java.util.Collection;
 import java.util.Set;
@@ -62,10 +63,12 @@ protected Set<MemberGroup> createInternalMemberGroups(Collection<? extends Membe
                                 + "check service definitions under META_INF.services folder. ");
                     } else {
                         for (DiscoveryStrategy discoveryStrategy : defaultDiscoveryService.getDiscoveryStrategies()) {
-                            checkNotNull(discoveryStrategy.getPartitionGroupStrategy());
-                            Iterable<MemberGroup> spiGroupsIterator =
-                                    discoveryStrategy.getPartitionGroupStrategy().getMemberGroups();
-                            for (MemberGroup group : spiGroupsIterator) {
+                            PartitionGroupStrategy groupStrategy = discoveryStrategy.getPartitionGroupStrategy(allMembers);
+                            if (groupStrategy == null) {
+                                groupStrategy = discoveryStrategy.getPartitionGroupStrategy();
+                            }
+                            checkNotNull(groupStrategy);
+                            for (MemberGroup group : groupStrategy.getMemberGroups()) {
                                 memberGroups.add(group);
                             }
                             return memberGroups;
diff --git a/hazelcast/src/main/java/com/hazelcast/spi/discovery/AbstractDiscoveryStrategy.java b/hazelcast/src/main/java/com/hazelcast/spi/discovery/AbstractDiscoveryStrategy.java
index 01702baea992..a0030dc44957 100644
--- a/hazelcast/src/main/java/com/hazelcast/spi/discovery/AbstractDiscoveryStrategy.java
+++ b/hazelcast/src/main/java/com/hazelcast/spi/discovery/AbstractDiscoveryStrategy.java
@@ -16,11 +16,13 @@
 
 package com.hazelcast.spi.discovery;
 
+import com.hazelcast.cluster.Member;
 import com.hazelcast.config.properties.PropertyDefinition;
 import com.hazelcast.logging.ILogger;
 import com.hazelcast.spi.partitiongroup.PartitionGroupStrategy;
 import com.hazelcast.internal.util.StringUtil;
 
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Map;
 
@@ -56,6 +58,11 @@ public PartitionGroupStrategy getPartitionGroupStrategy() {
         return null;
     }
 
+    @Override
+    public PartitionGroupStrategy getPartitionGroupStrategy(Collection<? extends Member> allMembers) {
+        return null;
+    }
+
     @Override
     public Map<String, String> discoverLocalMetadata() {
         return Collections.emptyMap();
diff --git a/hazelcast/src/main/java/com/hazelcast/spi/discovery/DiscoveryStrategy.java b/hazelcast/src/main/java/com/hazelcast/spi/discovery/DiscoveryStrategy.java
index 5d4fc5058049..0c98787d288f 100644
--- a/hazelcast/src/main/java/com/hazelcast/spi/discovery/DiscoveryStrategy.java
+++ b/hazelcast/src/main/java/com/hazelcast/spi/discovery/DiscoveryStrategy.java
@@ -19,6 +19,7 @@
 import com.hazelcast.cluster.Member;
 import com.hazelcast.spi.partitiongroup.PartitionGroupStrategy;
 
+import java.util.Collection;
 import java.util.Map;
 
 /**
@@ -65,6 +66,21 @@ public interface DiscoveryStrategy {
     void destroy();
 
     /**
+     * Returns a custom implementation of a {@link PartitionGroupStrategy} to override
+     * default behavior of zone aware backup strategies {@link com.hazelcast.spi.partitiongroup.PartitionGroupMetaData}
+     * or to provide a specific behavior in case the discovery environment does not provide
+     * information about the infrastructure to be used for automatic configuration.
+     * @param allMembers Current state of Cluster data members, excluding lite members
+     * @return a custom implementation of a <code>PartitionGroupStrategy</code> otherwise <code>null</code>
+     * in case of the default implementation is to be used
+     * @since 4.2.1
+     */
+    default PartitionGroupStrategy getPartitionGroupStrategy(Collection<? extends Member> allMembers) {
+        return null;
+    }
+
+    /**
+     * @deprecated - use the above method that takes allMember arguments
      * Returns a custom implementation of a {@link PartitionGroupStrategy} to override
      * default behavior of zone aware backup strategies {@link com.hazelcast.spi.partitiongroup.PartitionGroupMetaData}
      * or to provide a specific behavior in case the discovery environment does not provide
@@ -74,7 +90,10 @@ public interface DiscoveryStrategy {
      * in case of the default implementation is to be used
      * @since 3.7
      */
-    PartitionGroupStrategy getPartitionGroupStrategy();
+    @Deprecated
+    default PartitionGroupStrategy getPartitionGroupStrategy() {
+        return null;
+    }
 
     /**
      * Returns a map with discovered metadata provided by the runtime environment. Those information
diff --git a/hazelcast/src/test/java/com/hazelcast/spi/discovery/DiscoverySpiTest.java b/hazelcast/src/test/java/com/hazelcast/spi/discovery/DiscoverySpiTest.java
index 400b743ef4cf..b4043499d74d 100644
--- a/hazelcast/src/test/java/com/hazelcast/spi/discovery/DiscoverySpiTest.java
+++ b/hazelcast/src/test/java/com/hazelcast/spi/discovery/DiscoverySpiTest.java
@@ -267,7 +267,7 @@ public void testParsing() throws Exception {
     @Test
     public void testNodeStartup() {
         String xmlFileName = "test-hazelcast-discovery-spi.xml";
-        Config config = getDiscoverySPIConfig(xmlFileName);
+        Config config = getDiscoverySPIConfig(xmlFileName, false);
 
         try {
             final HazelcastInstance hazelcastInstance1 = Hazelcast.newHazelcastInstance(config);
@@ -376,7 +376,27 @@ public void testSPIAwareMemberGroupFactoryInvalidConfig() throws Exception {
     @Test
     public void testSPIAwareMemberGroupFactoryCreateMemberGroups() throws Exception {
         String xmlFileName = "test-hazelcast-discovery-spi-metadata.xml";
-        Config config = getDiscoverySPIConfig(xmlFileName);
+        Config config = getDiscoverySPIConfig(xmlFileName, false);
+        // we create this instance in order to fully create Node
+        HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance(config);
+        Node node = getNode(hazelcastInstance);
+        assertNotNull(node);
+
+        MemberGroupFactory groupFactory = new SPIAwareMemberGroupFactory(node.getDiscoveryService());
+        Collection<Member> members = createMembers();
+        Collection<MemberGroup> memberGroups = groupFactory.createMemberGroups(members);
+
+        assertEquals("Member Groups: " + String.valueOf(memberGroups), 2, memberGroups.size());
+        for (MemberGroup memberGroup : memberGroups) {
+            assertEquals("Member Group: " + String.valueOf(memberGroup), 2, memberGroup.size());
+        }
+        hazelcastInstance.shutdown();
+    }
+
+    @Test
+    public void testSPIAwareMemberGroupFactoryCreateMemberGroups_withDeprecated() throws Exception {
+        String xmlFileName = "test-hazelcast-discovery-spi-metadata.xml";
+        Config config = getDiscoverySPIConfig(xmlFileName, true);
         // we create this instance in order to fully create Node
         HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance(config);
         Node node = getNode(hazelcastInstance);
@@ -470,6 +490,23 @@ public void testMemberGroup_givenSPIMemberGroupIsActived_whenInstanceStarting_wo
         }
     }
 
+    @Test
+    public void noBreakingChangeInPartitionGroupStrategyAbstractClass() {
+        TestBreakingChangesDiscoveryStrategy strategy = new TestBreakingChangesDiscoveryStrategy();
+        assertNull(strategy.getPartitionGroupStrategy());
+    }
+
+    static class TestBreakingChangesDiscoveryStrategy extends AbstractDiscoveryStrategy {
+        TestBreakingChangesDiscoveryStrategy() {
+            super(null, Collections.<String, Comparable>emptyMap());
+        }
+
+        @Override
+        public Iterable<DiscoveryNode> discoverNodes() {
+            return null;
+        }
+    }
+
     @SuppressWarnings("unchecked")
     private static void setEnvironment(String key, String value) throws Exception {
         Class[] classes = Collections.class.getDeclaredClasses();
@@ -552,7 +589,7 @@ public void destroy() {
         }
 
         @Override
-        public PartitionGroupStrategy getPartitionGroupStrategy() {
+        public PartitionGroupStrategy getPartitionGroupStrategy(Collection<? extends Member> allMembers) {
             return null;
         }
 
@@ -562,6 +599,13 @@ public Map<String, String> discoverLocalMetadata() {
         }
     }
 
+    private static class DeprecatedTestDiscoveryStrategy extends TestDiscoveryStrategy {
+        @Override
+        public PartitionGroupStrategy getPartitionGroupStrategy() {
+            return null;
+        }
+    }
+
     public static class TestDiscoveryStrategyFactory implements DiscoveryStrategyFactory {
 
         private final Collection<PropertyDefinition> propertyDefinitions;
@@ -594,9 +638,9 @@ public Collection<PropertyDefinition> getConfigurationProperties() {
 
     public static class CollectingDiscoveryStrategyFactory implements DiscoveryStrategyFactory {
 
-        private final List<DiscoveryNode> discoveryNodes;
+        protected final List<DiscoveryNode> discoveryNodes;
 
-        private CollectingDiscoveryStrategyFactory(List<DiscoveryNode> discoveryNodes) {
+        CollectingDiscoveryStrategyFactory(List<DiscoveryNode> discoveryNodes) {
             this.discoveryNodes = discoveryNodes;
         }
 
@@ -616,6 +660,22 @@ public Collection<PropertyDefinition> getConfigurationProperties() {
         }
     }
 
+    private static class DeprecatedCollectingDiscoveryStrategyFactory extends CollectingDiscoveryStrategyFactory {
+        DeprecatedCollectingDiscoveryStrategyFactory(List<DiscoveryNode> discoveryNodes) {
+            super(discoveryNodes);
+        }
+
+        @Override
+        public Class<? extends DiscoveryStrategy> getDiscoveryStrategyType() {
+            return DeprecatedCollectingDiscoveryStrategy.class;
+        }
+
+        @Override
+        public DiscoveryStrategy newDiscoveryStrategy(DiscoveryNode node, ILogger logger, Map<String, Comparable> properties) {
+            return new DeprecatedCollectingDiscoveryStrategy(node, discoveryNodes, logger, properties);
+        }
+    }
+
     private static class CollectingDiscoveryStrategy extends AbstractDiscoveryStrategy {
 
         private final List<DiscoveryNode> discoveryNodes;
@@ -638,7 +698,7 @@ public void start() {
 
         // need to provide a custom impl
         @Override
-        public PartitionGroupStrategy getPartitionGroupStrategy() {
+        public PartitionGroupStrategy getPartitionGroupStrategy(Collection<? extends Member> allMembers) {
             return new SPIPartitionGroupStrategy();
         }
 
@@ -654,6 +714,18 @@ public void destroy() {
         }
     }
 
+    private static class DeprecatedCollectingDiscoveryStrategy extends CollectingDiscoveryStrategy {
+        DeprecatedCollectingDiscoveryStrategy(DiscoveryNode discoveryNode, List<DiscoveryNode> discoveryNodes, ILogger logger,
+                                              Map<String, Comparable> properties) {
+            super(discoveryNode, discoveryNodes, logger, properties);
+        }
+
+        @Override
+        public PartitionGroupStrategy getPartitionGroupStrategy() {
+            return new SPIPartitionGroupStrategy();
+        }
+    }
+
     public static class TestNodeFilter implements NodeFilter {
 
         private final List<DiscoveryNode> nodes = new ArrayList<DiscoveryNode>();
@@ -706,7 +778,7 @@ public Iterable<DiscoveryNode> discoverNodes() {
         }
 
         @Override
-        public PartitionGroupStrategy getPartitionGroupStrategy() {
+        public PartitionGroupStrategy getPartitionGroupStrategy(Collection<? extends Member> allMembers) {
             return new SPIPartitionGroupStrategy();
         }
 
@@ -718,6 +790,17 @@ public Map<String, String> discoverLocalMetadata() {
         }
     }
 
+    private static class DeprecatedMetadataProvidingDiscoveryStrategy extends MetadataProvidingDiscoveryStrategy {
+        DeprecatedMetadataProvidingDiscoveryStrategy(DiscoveryNode discoveryNode, ILogger logger, Map<String, Comparable> properties) {
+            super(discoveryNode, logger, properties);
+        }
+
+        @Override
+        public PartitionGroupStrategy getPartitionGroupStrategy() {
+            return new SPIPartitionGroupStrategy();
+        }
+    }
+
     private static class SPIPartitionGroupStrategy implements PartitionGroupStrategy {
 
         @Override
@@ -743,7 +826,7 @@ private static Collection<Member> createMembers() throws UnknownHostException {
         return members;
     }
 
-    private static Config getDiscoverySPIConfig(String xmlFileName) {
+    private static Config getDiscoverySPIConfig(String xmlFileName, boolean isDeprecated) {
         InputStream xmlResource = DiscoverySpiTest.class.getClassLoader().getResourceAsStream(xmlFileName);
         Config config = new XmlConfigBuilder(xmlResource).build();
         config.getNetworkConfig().setPort(50001);
@@ -753,7 +836,8 @@ private static Config getDiscoverySPIConfig(String xmlFileName) {
         interfaces.addInterface("127.0.0.1");
 
         List<DiscoveryNode> discoveryNodes = new CopyOnWriteArrayList<DiscoveryNode>();
-        DiscoveryStrategyFactory factory = new CollectingDiscoveryStrategyFactory(discoveryNodes);
+        DiscoveryStrategyFactory factory = isDeprecated ? new DeprecatedCollectingDiscoveryStrategyFactory(discoveryNodes)
+                : new CollectingDiscoveryStrategyFactory(discoveryNodes);
 
         DiscoveryConfig discoveryConfig = config.getNetworkConfig().getJoin().getDiscoveryConfig();
         discoveryConfig.getDiscoveryStrategyConfigs().clear();
diff --git a/hazelcast/src/test/resources/META-INF/services/com.hazelcast.spi.discovery.DiscoveryStrategyFactory b/hazelcast/src/test/resources/META-INF/services/com.hazelcast.spi.discovery.DiscoveryStrategyFactory
index a0fa05add21a..b458c59b943a 100644
--- a/hazelcast/src/test/resources/META-INF/services/com.hazelcast.spi.discovery.DiscoveryStrategyFactory
+++ b/hazelcast/src/test/resources/META-INF/services/com.hazelcast.spi.discovery.DiscoveryStrategyFactory
@@ -1,5 +1,7 @@
 com.hazelcast.spi.discovery.DiscoverySpiTest$TestDiscoveryStrategyFactory
+com.hazelcast.spi.discovery.DiscoverySpiTest$DeprecatedTestDiscoveryStrategyFactory
 com.hazelcast.spi.discovery.DiscoverySpiTest$MetadataProvidingDiscoveryStrategyFactory
+com.hazelcast.spi.discovery.DiscoverySpiTest$DeprecatedMetadataProvidingDiscoveryStrategyFactory
 com.hazelcast.spi.discovery.DiscoverySpiTest$ParametrizedDiscoveryStrategyFactory
 com.hazelcast.client.impl.spi.impl.discovery.ClientDiscoverySpiTest$TestDiscoveryStrategyFactory
 com.hazelcast.spi.discovery.impl.AutoDetectionDefaultDiscoveryServiceTest$TestDiscoveryStrategyFactory