How do I force JGroups which node to make coordinator?

Question

I'm looking for a way to force JGroups to use a specific server as the Coordinator, and if that server isn't present, elect a new Coordinator until that one specified rejoins the cluster and it takes over being the Coordinator.

In this case, we have some information we push in to the cluster by the Coordinator listening to a Topic for updates, however fetching & processing those updates can be resource intensive so we don't want it to server anything to the outside world. So in the load-balancer in front of the cluster, we have it set to not send to the coordinator. But because the Coordinator is elected randomly, we basically need to shut down the cluster until only the single machine is in there and then start the rest of the cluster back up.

Answer 1

Currently there is no way to do this. Jgroups has spent considerable time making sure that the coordinator can be any of the nodes in a group. All tasks that maintain and monitor the health of the group membership list are shared among all of the members in the group to make sure that the coordinator duties do not affect the performance of the coordinator too much. The standard GMS (Group MembershipService) protocol stack class is what is responsible for the coordinator selection. Currently it is just the first host in the view list.

To get this behavior, you are going to have to implement your own protocol stack. Interestingly, I've been working on a protocol stack for Jgroups which implements approximately what you are asking for however it is not ready for prime time.

Others may have taken a whack at this issue however. I would recommend posting on the jgroups mailing list and asking the same question.

Answer 2

You can set desired node to coordinator. : github sample

And i am add synchronization block for changes complete at all nodes and complete code:

public static final String GMS_DELTA_VIEW_FIELD_NAME = "use_delta_views";

/**
 * Change coordinator to {@code desiredCoordinator}. Must be invoked from coordinator.
 * @param desiredCoordinator
 * @return {@code true} if changes success, {@code false} overwise 
 */
boolean changeCoordinator(JChannel currentChannel, Address desiredCoordinator) {

    if(!Util.isCoordinator(currentChannel.getAddress)) {
        throw new RuntimeException("The current node is not coordinator.");
    }

    ArrayList<Address> newMembersOrder = Lists.newArrayList(currentView.getMembers());        

    // Switch desired node to first place
    Collections.swap(newMembersOrder, 0, newMembersOrder.indexOf(desiredCoordinator));        

    // Create new view
    long newId = currentView.getViewId().getId() + 1;
    View newView = new View(newMembersOrder.get(0), newId, newMembersOrder);

    GMS gms = (GMS)clusterChannel.getProtocolStack().findProtocol(GMS.class);
    CustomProtocol protocol = new CustomProtocol(newMembersOrder.stream()
            .filter(item -> !item.equals(currentChannel.getAddress()))
            .collect(Collectors.toSet()));

    boolean oldUseDeltaViews = (Boolean)gms.getValue(GMS_DELTA_VIEW_FIELD_NAME);
    try {
        // Disable using_delta_views at GMS
        gms.setValue(GMS_DELTA_VIEW_FIELD_NAME, false);

        // Insert custom protocol below GMS for synchronizing with VIEW_ACK events
        currentChannel.getProtocolStack().insertProtocolInStack(protocol, gms, ProtocolStack.BELOW);
        gms.castViewChange(newView, null, newMembersOrder);

        // Wait no more than 30 seconds to all VIEW_ACK responses
        if (!protocol.collector.waitForAllAcks(TimeUnit.SECONDS.toMillis(30))) {                
            return false;
        }

        return true;
    }
    finally {
        // Repair old state
        gms.setValue(GMS_DELTA_VIEW_FIELD_NAME, oldUseDeltaViews);
        currentChannel.getProtocolStack().removeProtocol(protocol);
    }
}

private class CustomProtocol extends Protocol implements UpHandler {

    AckCollector collector;

    public CustomProtocol(Collection<Address> waitedAddresses) {
        collector = new AckCollector(waitedAddresses);
    }

    @Override
    public Object up(Event evt) {

        if(evt.getType() == Event.MSG) {
            final Message msg=(Message)evt.getArg();
            GmsHeader hdr=(GmsHeader)msg.getHeader(proto_id);
            if(hdr != null && hdr.getType() == GmsHeader.VIEW_ACK) {                    
                collector.ack(msg.getSrc());
            }
        }

        return super.up(evt);
    }
}

Answer 3

Just stumbled across this post. There is a simple&standard way to do this in JGroups: [1]. It essentially let's user code control view generation.

[1] http://www.jgroups.org/manual4/index.html#MembershipChangePolicy