Skip to content

Commit 643ccbf

Browse files
GregDThomasakrherz
GregDThomas
authored and
akrherz
committed
Fix issue igniterealtime#19 - only change the NodeID if absolutely necessary.
1 parent 75b2bb1 commit 643ccbf

File tree

5 files changed

+92
-78
lines changed

5 files changed

+92
-78
lines changed

‎changelog.html

+1
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ <h1>
4949
<ul>
5050
<li>[<a href='https://github.com/igniterealtime/openfire-hazelcast-plugin/issues/13'>Issue #13</a>] - Less dramatic log messages when running a one-node cluster</li>
5151
<li>[<a href='https://github.com/igniterealtime/openfire-hazelcast-plugin/issues/16'>Issue #16</a>] - Expose host name of cluster members on the admin UI</li>
52+
<li>[<a href='https://github.com/igniterealtime/openfire-hazelcast-plugin/issues/19'>Issue #19</a>] - Stop changing the XMPPServer NodeID</li>
5253
</ul>
5354
<p><b>2.4.0</b> -- January 25, 2019</p>
5455

‎src/java/org/jivesoftware/openfire/plugin/HazelcastPlugin.java

+11
Original file line numberDiff line numberDiff line change
@@ -18,12 +18,15 @@
1818

1919
import java.io.File;
2020
import java.io.IOException;
21+
import java.nio.charset.StandardCharsets;
2122
import java.nio.file.Files;
2223
import java.nio.file.Path;
2324
import java.nio.file.Paths;
25+
import java.util.UUID;
2426

2527
import org.jivesoftware.openfire.XMPPServer;
2628
import org.jivesoftware.openfire.cluster.ClusterManager;
29+
import org.jivesoftware.openfire.cluster.NodeID;
2730
import org.jivesoftware.openfire.container.Plugin;
2831
import org.jivesoftware.openfire.container.PluginManager;
2932
import org.jivesoftware.openfire.container.PluginManagerListener;
@@ -45,6 +48,14 @@ public class HazelcastPlugin implements Plugin {
4548

4649
@Override
4750
public void initializePlugin(final PluginManager manager, final File pluginDirectory) {
51+
52+
if (XMPPServer.getInstance().getNodeID().equals(new byte[0])) {
53+
// TODO: (Greg 2019-03-14) Remove this when minServerVersion is 4.4.0 - that version does not require the node ID to be set
54+
final NodeID nodeID = NodeID.getInstance(UUID.randomUUID().toString().getBytes(StandardCharsets.UTF_8));
55+
LOGGER.info("Setting the XMPPServer node id once and for all to {}", nodeID);
56+
XMPPServer.getInstance().setNodeID(nodeID);
57+
}
58+
4859
LOGGER.info("Waiting for other plugins to initialize before initializing clustering");
4960
manager.addPluginManagerListener(new PluginManagerListener() {
5061
@Override

‎src/java/org/jivesoftware/openfire/plugin/util/cache/ClusterListener.java

+58-55
Original file line numberDiff line numberDiff line change
@@ -15,18 +15,20 @@
1515
*/
1616
package org.jivesoftware.openfire.plugin.util.cache;
1717

18-
import com.hazelcast.core.Cluster;
19-
import com.hazelcast.core.EntryEvent;
20-
import com.hazelcast.core.EntryEventType;
21-
import com.hazelcast.core.EntryListener;
22-
import com.hazelcast.core.LifecycleEvent;
23-
import com.hazelcast.core.LifecycleEvent.LifecycleState;
24-
import com.hazelcast.core.LifecycleListener;
25-
import com.hazelcast.core.MapEvent;
26-
import com.hazelcast.core.Member;
27-
import com.hazelcast.core.MemberAttributeEvent;
28-
import com.hazelcast.core.MembershipEvent;
29-
import com.hazelcast.core.MembershipListener;
18+
import java.nio.charset.StandardCharsets;
19+
import java.util.ArrayList;
20+
import java.util.Collection;
21+
import java.util.Collections;
22+
import java.util.HashMap;
23+
import java.util.HashSet;
24+
import java.util.Iterator;
25+
import java.util.List;
26+
import java.util.Map;
27+
import java.util.Set;
28+
import java.util.concurrent.ConcurrentHashMap;
29+
import java.util.concurrent.ConcurrentLinkedQueue;
30+
import java.util.concurrent.locks.Lock;
31+
3032
import org.jivesoftware.openfire.PacketException;
3133
import org.jivesoftware.openfire.RoutingTable;
3234
import org.jivesoftware.openfire.SessionManager;
@@ -53,19 +55,18 @@
5355
import org.xmpp.packet.JID;
5456
import org.xmpp.packet.Presence;
5557

56-
import java.nio.charset.StandardCharsets;
57-
import java.util.ArrayList;
58-
import java.util.Collection;
59-
import java.util.Collections;
60-
import java.util.HashMap;
61-
import java.util.HashSet;
62-
import java.util.Iterator;
63-
import java.util.List;
64-
import java.util.Map;
65-
import java.util.Set;
66-
import java.util.concurrent.ConcurrentHashMap;
67-
import java.util.concurrent.ConcurrentLinkedQueue;
68-
import java.util.concurrent.locks.Lock;
58+
import com.hazelcast.core.Cluster;
59+
import com.hazelcast.core.EntryEvent;
60+
import com.hazelcast.core.EntryEventType;
61+
import com.hazelcast.core.EntryListener;
62+
import com.hazelcast.core.LifecycleEvent;
63+
import com.hazelcast.core.LifecycleEvent.LifecycleState;
64+
import com.hazelcast.core.LifecycleListener;
65+
import com.hazelcast.core.MapEvent;
66+
import com.hazelcast.core.Member;
67+
import com.hazelcast.core.MemberAttributeEvent;
68+
import com.hazelcast.core.MembershipEvent;
69+
import com.hazelcast.core.MembershipListener;
6970

7071
/**
7172
* ClusterListener reacts to membership changes in the cluster. It takes care of cleaning up the state
@@ -113,7 +114,7 @@ public class ClusterListener implements MembershipListener, LifecycleListener {
113114
private final Map<Cache<?,?>, EntryListener> entryListeners = new HashMap<>();
114115

115116
private final Cluster cluster;
116-
private final Map<String, ClusterNodeInfo> clusterNodesInfo = new ConcurrentHashMap<>();
117+
private final Map<NodeID, ClusterNodeInfo> clusterNodesInfo = new ConcurrentHashMap<>();
117118

118119
/**
119120
* Flag that indicates if the listener has done all clean up work when noticed that the
@@ -130,7 +131,7 @@ public class ClusterListener implements MembershipListener, LifecycleListener {
130131

131132
this.cluster = cluster;
132133
for (Member member : cluster.getMembers()) {
133-
clusterNodesInfo.put(member.getUuid(),
134+
clusterNodesInfo.put(ClusteredCacheFactory.getNodeID(member),
134135
new HazelcastClusterNodeInfo(member, cluster.getClusterTime()));
135136
}
136137

@@ -415,7 +416,7 @@ private class DirectedPresenceListener implements EntryListener<String, Collecti
415416

416417
@Override
417418
public void entryAdded(EntryEvent<String, Collection<DirectedPresence>> event) {
418-
byte[] nodeID = event.getMember().getUuid().getBytes(StandardCharsets.UTF_8);
419+
final NodeID nodeID = ClusteredCacheFactory.getNodeID(event.getMember());
419420
// Ignore events originated from this JVM
420421
if (!XMPPServer.getInstance().getNodeID().equals(nodeID)) {
421422
// Check if the directed presence was sent to an entity hosted by this JVM
@@ -429,10 +430,10 @@ public void entryAdded(EntryEvent<String, Collection<DirectedPresence>> event) {
429430
}
430431
}
431432
if (!handlers.isEmpty()) {
432-
Map<String, Collection<String>> senders = nodePresences.get(NodeID.getInstance(nodeID));
433+
Map<String, Collection<String>> senders = nodePresences.get(nodeID);
433434
if (senders == null) {
434435
senders = new ConcurrentHashMap<>();
435-
nodePresences.put(NodeID.getInstance(nodeID), senders);
436+
nodePresences.put(nodeID, senders);
436437
}
437438
senders.put(sender, handlers);
438439
}
@@ -441,7 +442,7 @@ public void entryAdded(EntryEvent<String, Collection<DirectedPresence>> event) {
441442

442443
@Override
443444
public void entryUpdated(EntryEvent<String, Collection<DirectedPresence>> event) {
444-
byte[] nodeID = event.getMember().getUuid().getBytes(StandardCharsets.UTF_8);
445+
final NodeID nodeID = ClusteredCacheFactory.getNodeID(event.getMember());
445446
// Ignore events originated from this JVM
446447
if (!XMPPServer.getInstance().getNodeID().equals(nodeID)) {
447448
// Check if the directed presence was sent to an entity hosted by this JVM
@@ -454,10 +455,10 @@ public void entryUpdated(EntryEvent<String, Collection<DirectedPresence>> event)
454455
handlers.addAll(getReceivers(event, handler));
455456
}
456457
}
457-
Map<String, Collection<String>> senders = nodePresences.get(NodeID.getInstance(nodeID));
458+
Map<String, Collection<String>> senders = nodePresences.get(nodeID);
458459
if (senders == null) {
459460
senders = new ConcurrentHashMap<>();
460-
nodePresences.put(NodeID.getInstance(nodeID), senders);
461+
nodePresences.put(nodeID, senders);
461462
}
462463
if (!handlers.isEmpty()) {
463464
senders.put(sender, handlers);
@@ -475,10 +476,10 @@ public void entryRemoved(EntryEvent<String, Collection<DirectedPresence>> event)
475476
// Nothing to remove
476477
return;
477478
}
478-
byte[] nodeID = event.getMember().getUuid().getBytes(StandardCharsets.UTF_8);
479+
final NodeID nodeID = ClusteredCacheFactory.getNodeID(event.getMember());
479480
if (!XMPPServer.getInstance().getNodeID().equals(nodeID)) {
480481
String sender = event.getKey();
481-
nodePresences.get(NodeID.getInstance(nodeID)).remove(sender);
482+
nodePresences.get(nodeID).remove(sender);
482483
}
483484
}
484485

@@ -509,7 +510,7 @@ public void entryEvicted(EntryEvent<String, Collection<DirectedPresence>> event)
509510
}
510511

511512
private void mapClearedOrEvicted(MapEvent event) {
512-
NodeID nodeID = NodeID.getInstance(event.getMember().getUuid().getBytes(StandardCharsets.UTF_8));
513+
final NodeID nodeID = ClusteredCacheFactory.getNodeID(event.getMember());
513514
// ignore events which were triggered by this node
514515
if (!XMPPServer.getInstance().getNodeID().equals(nodeID)) {
515516
nodePresences.get(nodeID).clear();
@@ -579,7 +580,7 @@ public void entryEvicted(EntryEvent<String, Set<NodeID>> event) {
579580
}
580581

581582
private void mapClearedOrEvicted(MapEvent event) {
582-
NodeID nodeID = NodeID.getInstance(event.getMember().getUuid().getBytes(StandardCharsets.UTF_8));
583+
final NodeID nodeID = ClusteredCacheFactory.getNodeID(event.getMember());
583584
// ignore events which were triggered by this node
584585
if (!XMPPServer.getInstance().getNodeID().equals(nodeID)) {
585586
Set<String> sessionJIDs = lookupJIDList(nodeID, componentsCache.getName());
@@ -634,8 +635,8 @@ synchronized void joinCluster() {
634635
if (seniorClusterMember) {
635636
ClusterManager.fireMarkedAsSeniorClusterMember();
636637
}
637-
logger.info("Joined cluster as node: " + cluster.getLocalMember().getUuid() + ". Senior Member: " +
638-
(seniorClusterMember ? "YES" : "NO"));
638+
logger.info("Joined cluster. XMPPServer node={}, Hazelcast UUID={}, seniorClusterMember={}",
639+
new Object[]{ClusteredCacheFactory.getNodeID(cluster.getLocalMember()), cluster.getLocalMember().getUuid(), seniorClusterMember});
639640
done = false;
640641
}
641642

@@ -650,6 +651,7 @@ private synchronized void leaveCluster() {
650651
return;
651652
}
652653
clusterMember = false;
654+
final boolean wasSeniorClusterMember = seniorClusterMember;
653655
seniorClusterMember = false;
654656
// Clean up all traces. This will set all remote sessions as unavailable
655657
List<NodeID> nodeIDs = new ArrayList<>(nodeSessions.keySet());
@@ -670,54 +672,55 @@ private synchronized void leaveCluster() {
670672
// At this point c2s sessions are gone from the routing table so we can identify expired sessions
671673
XMPPServer.getInstance().getPresenceUpdateHandler().removedExpiredPresences();
672674
}
673-
logger.info("Left cluster as node: " + cluster.getLocalMember().getUuid());
675+
logger.info("Left cluster. XMPPServer node={}, Hazelcast UUID={}, wasSeniorClusterMember={}",
676+
new Object[]{ClusteredCacheFactory.getNodeID(cluster.getLocalMember()), cluster.getLocalMember().getUuid(), wasSeniorClusterMember});
674677
done = true;
675678
}
676679

677680
@Override
678681
public void memberAdded(MembershipEvent event) {
679682
// local member only
683+
final NodeID nodeID = ClusteredCacheFactory.getNodeID(event.getMember());
680684
if (event.getMember().localMember()) { // We left and re-joined the cluster
681685
joinCluster();
682686
} else {
683-
nodePresences.put(NodeID.getInstance(event.getMember().getUuid().getBytes(StandardCharsets.UTF_8)),
684-
new ConcurrentHashMap<>());
687+
nodePresences.put(nodeID, new ConcurrentHashMap<>());
685688
// Trigger event that a new node has joined the cluster
686-
ClusterManager.fireJoinedCluster(event.getMember().getUuid().getBytes(StandardCharsets.UTF_8), true);
689+
ClusterManager.fireJoinedCluster(nodeID.toByteArray(), true);
687690
}
688-
clusterNodesInfo.put(event.getMember().getUuid(),
691+
clusterNodesInfo.put(nodeID,
689692
new HazelcastClusterNodeInfo(event.getMember(), cluster.getClusterTime()));
690693
}
691694

692695
@Override
693696
public void memberRemoved(MembershipEvent event) {
694-
byte[] nodeID = event.getMember().getUuid().getBytes(StandardCharsets.UTF_8);
697+
final NodeID nodeID = ClusteredCacheFactory.getNodeID(event.getMember());
695698

696699
if (event.getMember().localMember()) {
697-
logger.info("Leaving cluster: " + new String(nodeID, StandardCharsets.UTF_8));
700+
logger.info("Leaving cluster: " + nodeID);
698701
// This node may have realized that it got kicked out of the cluster
699702
leaveCluster();
700703
} else {
701704
// Trigger event that a node left the cluster
702-
ClusterManager.fireLeftCluster(nodeID);
705+
ClusterManager.fireLeftCluster(nodeID.toByteArray());
703706

704707
// Clean up directed presences sent from entities hosted in the leaving node to local entities
705708
// Clean up directed presences sent to entities hosted in the leaving node from local entities
706-
cleanupDirectedPresences(NodeID.getInstance(nodeID));
709+
cleanupDirectedPresences(nodeID);
707710

708711
if (!seniorClusterMember && isSeniorClusterMember()) {
709712
seniorClusterMember = true;
710713
ClusterManager.fireMarkedAsSeniorClusterMember();
711714
}
712-
cleanupNode(NodeID.getInstance(nodeID));
715+
cleanupNode(nodeID);
713716

714717
// Remove traces of directed presences sent from local entities to handlers that no longer exist.
715718
// At this point c2s sessions are gone from the routing table so we can identify expired sessions
716719
XMPPServer.getInstance().getPresenceUpdateHandler().removedExpiredPresences();
717720
}
718721
// Delete nodeID instance (release from memory)
719-
NodeID.deleteInstance(nodeID);
720-
clusterNodesInfo.remove(event.getMember().getUuid());
722+
NodeID.deleteInstance(nodeID.toByteArray());
723+
clusterNodesInfo.remove(nodeID);
721724
}
722725

723726
@SuppressWarnings("WeakerAccess")
@@ -736,8 +739,8 @@ public void stateChanged(LifecycleEvent event) {
736739

737740
@Override
738741
public void memberAttributeChanged(MemberAttributeEvent event) {
739-
ClusterNodeInfo priorNodeInfo = clusterNodesInfo.get(event.getMember().getUuid());
740-
clusterNodesInfo.put(event.getMember().getUuid(),
742+
ClusterNodeInfo priorNodeInfo = clusterNodesInfo.get(ClusteredCacheFactory.getNodeID(event.getMember()));
743+
clusterNodesInfo.put(ClusteredCacheFactory.getNodeID(event.getMember()),
741744
new HazelcastClusterNodeInfo(event.getMember(), priorNodeInfo.getJoinedTime()));
742745
}
743746

@@ -766,7 +769,7 @@ public void entryEvicted(EntryEvent<DomainPair, byte[]> event) {
766769
}
767770

768771
private void handleEntryEvent(EntryEvent<DomainPair, byte[]> event, boolean removal) {
769-
NodeID nodeID = NodeID.getInstance(event.getMember().getUuid().getBytes(StandardCharsets.UTF_8));
772+
NodeID nodeID = ClusteredCacheFactory.getNodeID(event.getMember());
770773
// ignore events which were triggered by this node
771774
if (!XMPPServer.getInstance().getNodeID().equals(nodeID)) {
772775
Set<DomainPair> sessionJIDS = nodeRoutes.get(nodeID);
@@ -782,7 +785,7 @@ private void handleEntryEvent(EntryEvent<DomainPair, byte[]> event, boolean remo
782785
}
783786

784787
private void handleMapEvent(MapEvent event) {
785-
NodeID nodeID = NodeID.getInstance(event.getMember().getUuid().getBytes(StandardCharsets.UTF_8));
788+
NodeID nodeID = ClusteredCacheFactory.getNodeID(event.getMember());
786789
// ignore events which were triggered by this node
787790
if (!XMPPServer.getInstance().getNodeID().equals(nodeID)) {
788791
Set<DomainPair> sessionJIDS = nodeRoutes.get(nodeID);

0 commit comments

Comments
 (0)