Demote local data member

hazelcast/src/main/java/com/hazelcast/auditlog/AuditlogTypeIds.java

@@ -86,6 +86,10 @@ public final class AuditlogTypeIds {
      * Event type ID: Clusters merged.
      */
     public static final String CLUSTER_MERGE = "HZ-0606";
+    /**
+     * Event type ID: Data member demoted.
+     */
+    public static final String CLUSTER_DEMOTE_MEMBER = "HZ-0607";
 
     // Member events
     /**

hazelcast/src/main/java/com/hazelcast/client/impl/proxy/ClientClusterProxy.java

@@ -131,4 +131,9 @@ public void changeClusterVersion(@Nonnull Version version, @Nonnull TransactionO
     public void promoteLocalLiteMember() {
         throw new UnsupportedOperationException();
     }
+
+    @Override
+    public void demoteLocalDataMember() {
+        throw new UnsupportedOperationException();
+    }
 }

hazelcast/src/main/java/com/hazelcast/cluster/Cluster.java

@@ -112,6 +112,21 @@ public interface Cluster {
      */
     void promoteLocalLiteMember();
 
+    /**
+     * Demotes the local data member to a lite member.
+     * When this method returns, both {@link #getLocalMember()} and {@link #getMembers()}
+     * reflect the demotion.
+     * <p>
+     * Supported only for members of the cluster, clients will throw a {@code UnsupportedOperationException}.
+     *
+     * @throws IllegalStateException when member is not a data member or mastership claim is in progress
+     *                               or local member cannot be identified as a member of the cluster
+     *                               or cluster state doesn't allow migrations/repartitioning
+     *                               or the member is the only data member in the cluster.
+     * @since 5.4
+     */
+    void demoteLocalDataMember();
+
     /**
      * Returns the cluster-wide time in milliseconds.
      * <p>

hazelcast/src/main/java/com/hazelcast/internal/cluster/ClusterService.java

@@ -133,7 +133,8 @@ public interface ClusterService extends CoreService, Cluster {
      * Gets the local member instance.
      * <p>
      * The returned value will never be null, but it may change when local lite member is promoted to a data member
-     * via {@link #promoteLocalLiteMember()}
+     * via {@link #promoteLocalLiteMember()} or when local data member is demoted to a lite member via
+     * {@link #demoteLocalDataMember()}
      * or when this member merges to a new cluster after split-brain detected. Returned value should not be
      * cached but instead this method should be called each time when local member is needed.
      *

hazelcast/src/main/java/com/hazelcast/internal/cluster/impl/ClusterDataSerializerHook.java

@@ -24,6 +24,7 @@
 import com.hazelcast.internal.cluster.impl.operations.BeforeJoinCheckFailureOp;
 import com.hazelcast.internal.cluster.impl.operations.CommitClusterStateOp;
 import com.hazelcast.internal.cluster.impl.operations.ConfigMismatchOp;
+import com.hazelcast.internal.cluster.impl.operations.DemoteDataMemberOp;
 import com.hazelcast.internal.cluster.impl.operations.ExplicitSuspicionOp;
 import com.hazelcast.internal.cluster.impl.operations.FetchMembersViewOp;
 import com.hazelcast.internal.cluster.impl.operations.FinalizeJoinOp;
@@ -100,8 +101,11 @@ public final class ClusterDataSerializerHook implements DataSerializerHook {
     public static final int HEARTBEAT_COMPLAINT = 38;
     public static final int PROMOTE_LITE_MEMBER = 39;
     public static final int VECTOR_CLOCK = 40;
+    public static final int DEMOTE_DATA_MEMBER = 41;
+    public static final int MEMBERS_VIEW_RESPONSE = 42;
 
-    static final int LEN = VECTOR_CLOCK + 1;
+
+    static final int LEN = MEMBERS_VIEW_RESPONSE + 1;
 
     @Override
     public int getFactoryId() {
@@ -151,6 +155,8 @@ public DataSerializableFactory createFactory() {
         constructors[MEMBERS_VIEW_METADATA] = arg -> new MembersViewMetadata();
         constructors[HEARTBEAT_COMPLAINT] = arg -> new HeartbeatComplaintOp();
         constructors[PROMOTE_LITE_MEMBER] = arg -> new PromoteLiteMemberOp();
+        constructors[DEMOTE_DATA_MEMBER] = arg -> new DemoteDataMemberOp();
+        constructors[MEMBERS_VIEW_RESPONSE] = arg -> new MembersViewResponse();
         constructors[VECTOR_CLOCK] = arg -> new VectorClock();
         constructors[ENDPOINT_QUALIFIER] = arg -> new EndpointQualifier();
         return new ArrayDataSerializableFactory(constructors);

hazelcast/src/main/java/com/hazelcast/internal/cluster/impl/ClusterServiceImpl.java

@@ -32,6 +32,8 @@
 import com.hazelcast.instance.impl.LifecycleServiceImpl;
 import com.hazelcast.instance.impl.Node;
 import com.hazelcast.internal.cluster.ClusterService;
+import com.hazelcast.internal.cluster.Versions;
+import com.hazelcast.internal.cluster.impl.operations.DemoteDataMemberOp;
 import com.hazelcast.internal.cluster.impl.operations.ExplicitSuspicionOp;
 import com.hazelcast.internal.cluster.impl.operations.OnJoinOp;
 import com.hazelcast.internal.cluster.impl.operations.PromoteLiteMemberOp;
@@ -90,6 +92,7 @@
 import static com.hazelcast.internal.util.Preconditions.checkNotNull;
 import static com.hazelcast.internal.util.Preconditions.checkTrue;
 import static java.lang.String.format;
+import static java.util.concurrent.TimeUnit.SECONDS;
 
 @SuppressWarnings({"checkstyle:methodcount", "checkstyle:classdataabstractioncoupling", "checkstyle:classfanoutcomplexity"})
 public class ClusterServiceImpl implements ClusterService, ConnectionListener, ManagedService,
@@ -1046,13 +1049,13 @@ public void promoteLocalLiteMember() {
             }
 
             MemberImpl localMemberInMemberList = membershipManager.getMember(member.getAddress());
-            boolean result = localMemberInMemberList.isLiteMember();
+            boolean isStillLiteMember = localMemberInMemberList.isLiteMember();
             node.getNodeExtension().getAuditlogService().eventBuilder(AuditlogTypeIds.CLUSTER_PROMOTE_MEMBER)
                 .message("Promotion of the lite member")
-                .addParameter("success", result)
+                .addParameter("success", !isStillLiteMember)
                 .addParameter("address", node.getThisAddress())
                 .log();
-            if (result) {
+            if (isStillLiteMember) {
                 throw new IllegalStateException("Cannot promote to data member! Previous master was: " + master.getAddress()
                         + ", Current master is: " + getMasterAddress());
             }
@@ -1078,6 +1081,72 @@ MemberImpl promoteAndGetLocalMember() {
         return localMember;
     }
 
+    MemberImpl demoteAndGetLocalMember() {
+        MemberImpl member = getLocalMember();
+        assert !member.isLiteMember() : "Local member is not data member!";
+        assert clusterServiceLock.isHeldByCurrentThread() : "Called without holding cluster service lock!";
+
+        localMember = new MemberImpl.Builder(member.getAddressMap())
+                .version(member.getVersion())
+                .localMember(true)
+                .uuid(member.getUuid())
+                .attributes(member.getAttributes())
+                .memberListJoinVersion(member.getMemberListJoinVersion())
+                .instance(node.hazelcastInstance)
+                .liteMember(true)
+                .build();
+        node.loggingService.setThisMember(localMember);
+        return localMember;
+    }
+
+
+    @Override
+    public void demoteLocalDataMember() {
+
+        if (getClusterVersion().isUnknownOrLessThan(Versions.V5_4)) {
+            throw new UnsupportedOperationException("demoteLocalDataMember requires cluster version 5.4 or greater");
+        }
+
+        MemberImpl member = getLocalMember();
+        if (member.isLiteMember()) {
+            throw new IllegalStateException(member + " is not a data member!");
+        }
+
+        MemberImpl master = getMasterMember();
+
+        long maxWaitSeconds = node.getProperties().getSeconds(ClusterProperty.DEMOTE_MAX_WAIT);
+        if (!nodeEngine.getPartitionService().onDemote(maxWaitSeconds, SECONDS)) {
+            throw new IllegalStateException("Cannot demote to lite member! Previous master was: " + master.getAddress()
+                    + ", Current master is: " + getMasterAddress() + ". Cluster state is " + getClusterState());
+        }
+
+        DemoteDataMemberOp op = new DemoteDataMemberOp();
+        op.setCallerUuid(member.getUuid());
+        InvocationFuture<MembersViewResponse> future = nodeEngine.getOperationService().invokeOnMaster(SERVICE_NAME, op);
+        MembersViewResponse response = future.joinInternal();
+
+        clusterServiceLock.lock();
+        try {
+            if (!node.isMaster()) {
+                updateMembers(response.getMembersView(), response.getMemberAddress(), response.getMemberUuid(), getThisUuid());
+            }
+
+            MemberImpl localMemberInMemberList = membershipManager.getMember(member.getAddress());
+            boolean isNowLiteMember = localMemberInMemberList.isLiteMember();
+            node.getNodeExtension().getAuditlogService().eventBuilder(AuditlogTypeIds.CLUSTER_DEMOTE_MEMBER)
+                    .message("Demotion of the data member")
+                    .addParameter("success", isNowLiteMember)
+                    .addParameter("address", node.getThisAddress())
+                    .log();
+            if (!isNowLiteMember) {
+                throw new IllegalStateException("Cannot demote to lite member! Previous master was: " + master.getAddress()
+                        + ", Current master is: " + getMasterAddress());
+            }
+        } finally {
+            clusterServiceLock.unlock();
+        }
+    }
+
     @Override
     public int getMemberListVersion() {
         return membershipManager.getMemberListVersion();

hazelcast/src/main/java/com/hazelcast/internal/cluster/impl/MembersViewResponse.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright (c) 2008-2023, Hazelcast, Inc. All Rights Reserved.
+ *
+ * Licensed 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 com.hazelcast.internal.cluster.impl;
+
+import com.hazelcast.cluster.Address;
+import com.hazelcast.internal.util.UUIDSerializationUtil;
+import com.hazelcast.nio.ObjectDataInput;
+import com.hazelcast.nio.ObjectDataOutput;
+import com.hazelcast.nio.serialization.IdentifiedDataSerializable;
+
+import java.io.IOException;
+import java.util.UUID;
+
+public class MembersViewResponse implements IdentifiedDataSerializable {
+
+    private Address memberAddress;
+
+    private UUID memberUuid;
+
+    private MembersView membersView;
+
+    public MembersViewResponse() {
+    }
+
+    public MembersViewResponse(Address memberAddress, UUID memberUuid, MembersView membersView) {
+        this.memberAddress = memberAddress;
+        this.memberUuid = memberUuid;
+        this.membersView = membersView;
+    }
+
+
+    public Address getMemberAddress() {
+        return memberAddress;
+    }
+
+    public UUID getMemberUuid() {
+        return memberUuid;
+    }
+
+    public MembersView getMembersView() {
+        return membersView;
+    }
+
+    @Override
+    public void writeData(ObjectDataOutput out) throws IOException {
+        out.writeObject(memberAddress);
+        UUIDSerializationUtil.writeUUID(out, memberUuid);
+        out.writeObject(membersView);
+    }
+
+    @Override
+    public void readData(ObjectDataInput in) throws IOException {
+        memberAddress = in.readObject();
+        memberUuid = UUIDSerializationUtil.readUUID(in);
+        membersView = in.readObject();
+    }
+
+    @Override
+    public int getFactoryId() {
+        return ClusterDataSerializerHook.F_ID;
+    }
+
+    @Override
+    public int getClassId() {
+        return ClusterDataSerializerHook.MEMBERS_VIEW_RESPONSE;
+    }
+}