Add nested cluster multicast/membership configurations.

agent/src/main/java/io/atomix/agent/AtomixAgent.java

@@ -95,10 +95,14 @@ public static void main(String[] args) throws Exception {
         .action(new StoreTrueArgumentAction())
         .setDefault(false)
         .help("Enables multicast discovery. Note that the network must support multicast for this feature to work.");
-    parser.addArgument("--multicast-address")
-        .type(addressArgumentType)
-        .metavar("IP:PORT")
-        .help("Sets the multicast discovery address. Defaults to 230.0.0.1:54321");
+    parser.addArgument("--multicast-group")
+        .type(String.class)
+        .metavar("IP")
+        .help("Sets the multicast group. Defaults to 230.0.0.1");
+    parser.addArgument("--multicast-port")
+        .type(Integer.class)
+        .metavar("PORT")
+        .help("Sets the multicast port. Defaults to 54321");
     parser.addArgument("--http-port", "-p")
         .type(Integer.class)
         .metavar("PORT")
@@ -122,7 +126,8 @@ public static void main(String[] args) throws Exception {
     final String zone = namespace.getString("zone");
     final List<Node> bootstrap = namespace.getList("bootstrap");
     final boolean multicastEnabled = namespace.getBoolean("multicast");
-    final Address multicastAddress = namespace.get("multicast_address");
+    final String multicastGroup = namespace.get("multicast_group");
+    final Integer multicastPort = namespace.get("multicast_port");
     final Integer httpPort = namespace.getInt("http_port");
 
     // If a configuration was provided, merge the configuration's member information with the provided command line arguments.
@@ -156,8 +161,13 @@ public static void main(String[] args) throws Exception {
     }
 
     if (multicastEnabled) {
-      config.getClusterConfig().setMulticastEnabled(true);
-      config.getClusterConfig().setMulticastAddress(multicastAddress);
+      config.getClusterConfig().getMulticastConfig().setEnabled(true);
+      if (multicastGroup != null) {
+        config.getClusterConfig().getMulticastConfig().setGroup(multicastGroup);
+      }
+      if (multicastPort != null) {
+        config.getClusterConfig().getMulticastConfig().setPort(multicastPort);
+      }
       if (bootstrap == null || bootstrap.isEmpty()) {
         config.getClusterConfig().setDiscoveryConfig(new MulticastDiscoveryProvider.Config());
       }

agent/src/test/resources/test.conf

@@ -1,6 +1,6 @@
 cluster {
   name: test
-  multicast-enabled: true
+  multicast.enabled: true
 }
 
 managementGroup {

cluster/src/main/java/io/atomix/cluster/AtomixCluster.java

@@ -321,8 +321,11 @@ protected static ManagedMessagingService buildMessagingService(ClusterConfig con
   protected static ManagedBroadcastService buildBroadcastService(ClusterConfig config) {
     return NettyBroadcastService.builder()
         .withLocalAddress(config.getAddress())
-        .withGroupAddress(config.getMulticastAddress())
-        .withEnabled(config.isMulticastEnabled())
+        .withGroupAddress(new Address(
+            config.getMulticastConfig().getGroup().getHostName(),
+            config.getMulticastConfig().getPort(),
+            config.getMulticastConfig().getGroup()))
+        .withEnabled(config.getMulticastConfig().isEnabled())
         .build();
   }
 
@@ -331,11 +334,11 @@ protected static ManagedBroadcastService buildBroadcastService(ClusterConfig con
    */
   @SuppressWarnings("unchecked")
   protected static NodeDiscoveryProvider buildLocationProvider(ClusterConfig config) {
-    NodeDiscoveryProvider.Config locationProviderConfig = config.getDiscoveryConfig();
-    if (locationProviderConfig != null) {
-      return locationProviderConfig.getType().newProvider(locationProviderConfig);
+    NodeDiscoveryProvider.Config discoveryProviderConfig = config.getDiscoveryConfig();
+    if (discoveryProviderConfig != null) {
+      return discoveryProviderConfig.getType().newProvider(discoveryProviderConfig);
     }
-    if (config.isMulticastEnabled()) {
+    if (config.getMulticastConfig().isEnabled()) {
       return new MulticastDiscoveryProvider(new MulticastDiscoveryProvider.Config());
     } else {
       return new BootstrapDiscoveryProvider(Collections.emptyList());
@@ -362,7 +365,7 @@ protected static ManagedClusterMembershipService buildClusterMembershipService(
         localMember,
         new DefaultNodeDiscoveryService(bootstrapService, localMember, discoveryProvider),
         bootstrapService,
-        config);
+        config.getMembershipConfig());
   }
 
   /**
@@ -547,7 +550,7 @@ public Builder withMulticastEnabled() {
      * @return the Atomix builder
      */
     public Builder withMulticastEnabled(boolean multicastEnabled) {
-      config.setMulticastEnabled(multicastEnabled);
+      config.getMulticastConfig().setEnabled(multicastEnabled);
       return this;
     }
 
@@ -558,7 +561,8 @@ public Builder withMulticastEnabled(boolean multicastEnabled) {
      * @return the Atomix builder
      */
     public Builder withMulticastAddress(Address address) {
-      config.setMulticastAddress(address);
+      config.getMulticastConfig().setGroup(address.address());
+      config.getMulticastConfig().setPort(address.port());
       return this;
     }
 
@@ -569,7 +573,7 @@ public Builder withMulticastAddress(Address address) {
      * @return the Atomix builder
      */
     public Builder setBroadcastInterval(Duration interval) {
-      config.setBroadcastInterval((int) interval.toMillis());
+      config.getMembershipConfig().setBroadcastInterval((int) interval.toMillis());
       return this;
     }
 
@@ -580,7 +584,7 @@ public Builder setBroadcastInterval(Duration interval) {
      * @return the Atomix builder
      */
     public Builder setReachabilityThreshold(int threshold) {
-      config.setReachabilityThreshold(threshold);
+      config.getMembershipConfig().setReachabilityThreshold(threshold);
       return this;
     }
 
@@ -591,7 +595,7 @@ public Builder setReachabilityThreshold(int threshold) {
      * @return the Atomix builder
      */
     public Builder withReachabilityTimeout(Duration timeout) {
-      config.setReachabilityTimeout((int) timeout.toMillis());
+      config.getMembershipConfig().setReachabilityTimeout((int) timeout.toMillis());
       return this;
     }
 

cluster/src/main/java/io/atomix/cluster/ClusterConfig.java

@@ -17,7 +17,6 @@
 
 import io.atomix.utils.config.Config;
 import io.atomix.utils.net.Address;
-import io.atomix.utils.net.MalformedAddressException;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -29,11 +28,6 @@
  */
 public class ClusterConfig implements Config {
   private static final String DEFAULT_CLUSTER_NAME = "atomix";
-  private static final String DEFAULT_MULTICAST_IP = "230.0.0.1";
-  private static final int DEFAULT_MULTICAST_PORT = 54321;
-  private static final int DEFAULT_BROADCAST_INTERVAL = 100;
-  private static final int DEFAULT_REACHABILITY_TIMEOUT = 10000;
-  private static final int DEFAULT_REACHABILITY_THRESHOLD = 10;
 
   private String clusterId = DEFAULT_CLUSTER_NAME;
   private MemberId memberId = MemberId.anonymous();
@@ -43,19 +37,8 @@ public class ClusterConfig implements Config {
   private String host;
   private Map<String, String> metadata = new HashMap<>();
   private NodeDiscoveryProvider.Config discoveryConfig;
-  private boolean multicastEnabled = false;
-  private Address multicastAddress;
-  private int broadcastInterval = DEFAULT_BROADCAST_INTERVAL;
-  private int reachabilityThreshold = DEFAULT_REACHABILITY_THRESHOLD;
-  private int reachabilityTimeout = DEFAULT_REACHABILITY_TIMEOUT;
-
-  public ClusterConfig() {
-    try {
-      multicastAddress = Address.from(DEFAULT_MULTICAST_IP, DEFAULT_MULTICAST_PORT);
-    } catch (MalformedAddressException e) {
-      multicastAddress = Address.from(DEFAULT_MULTICAST_PORT);
-    }
-  }
+  private MulticastConfig multicastConfig = new MulticastConfig();
+  private MembershipConfig membershipConfig = new MembershipConfig();
 
   /**
    * Returns the cluster identifier.
@@ -250,102 +233,42 @@ public ClusterConfig setDiscoveryConfig(NodeDiscoveryProvider.Config discoveryCo
   }
 
   /**
-   * Returns whether multicast is enabled.
-   *
-   * @return whether multicast is enabled
-   */
-  public boolean isMulticastEnabled() {
-    return multicastEnabled;
-  }
-
-  /**
-   * Sets whether multicast is enabled.
-   *
-   * @param multicastEnabled whether multicast is enabled
-   * @return the cluster configuration
-   */
-  public ClusterConfig setMulticastEnabled(boolean multicastEnabled) {
-    this.multicastEnabled = multicastEnabled;
-    return this;
-  }
-
-  /**
-   * Returns the multicast address.
-   *
-   * @return the multicast address
-   */
-  public Address getMulticastAddress() {
-    return multicastAddress;
-  }
-
-  /**
-   * Sets the multicast address.
-   *
-   * @param multicastAddress the multicast address
-   * @return the cluster configuration
-   */
-  public ClusterConfig setMulticastAddress(Address multicastAddress) {
-    this.multicastAddress = multicastAddress;
-    return this;
-  }
-
-  /**
-   * Returns the reachability broadcast interval.
-   *
-   * @return the reachability broadcast interval
-   */
-  public int getBroadcastInterval() {
-    return broadcastInterval;
-  }
-
-  /**
-   * Sets the reachability broadcast interval.
-   *
-   * @param broadcastInterval the reachability broadcast interval
-   * @return the cluster configuration
-   */
-  public ClusterConfig setBroadcastInterval(int broadcastInterval) {
-    this.broadcastInterval = broadcastInterval;
-    return this;
-  }
-
-  /**
-   * Returns the reachability failure detection threshold.
+   * Returns the multicast configuration.
    *
-   * @return the reachability failure detection threshold
+   * @return the multicast configuration
    */
-  public int getReachabilityThreshold() {
-    return reachabilityThreshold;
+  public MulticastConfig getMulticastConfig() {
+    return multicastConfig;
   }
 
   /**
-   * Sets the reachability failure detection threshold.
+   * Sets the multicast configuration.
    *
-   * @param reachabilityThreshold the reachability failure detection threshold
+   * @param multicastConfig the multicast configuration
    * @return the cluster configuration
    */
-  public ClusterConfig setReachabilityThreshold(int reachabilityThreshold) {
-    this.reachabilityThreshold = reachabilityThreshold;
+  public ClusterConfig setMulticastConfig(MulticastConfig multicastConfig) {
+    this.multicastConfig = checkNotNull(multicastConfig);
     return this;
   }
 
   /**
-   * Returns the reachability failure timeout.
+   * Returns the cluster membership configuration.
    *
-   * @return the reachability failure timeout
+   * @return the cluster membership configuration
    */
-  public int getReachabilityTimeout() {
-    return reachabilityTimeout;
+  public MembershipConfig getMembershipConfig() {
+    return membershipConfig;
   }
 
   /**
-   * Sets the reachability failure timeout.
+   * Sets the cluster membership configuration.
    *
-   * @param reachabilityTimeout the reachability failure timeout
+   * @param membershipConfig the cluster membership configuration
    * @return the cluster configuration
    */
-  public ClusterConfig setReachabilityTimeout(int reachabilityTimeout) {
-    this.reachabilityTimeout = reachabilityTimeout;
+  public ClusterConfig setMembershipConfig(MembershipConfig membershipConfig) {
+    this.membershipConfig = membershipConfig;
     return this;
   }
 }

cluster/src/main/java/io/atomix/cluster/MembershipConfig.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+ *
+ * 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 io.atomix.cluster;
+
+import io.atomix.utils.config.Config;
+
+/**
+ * Cluster membership configuration.
+ */
+public class MembershipConfig implements Config {
+  private static final int DEFAULT_BROADCAST_INTERVAL = 100;
+  private static final int DEFAULT_REACHABILITY_TIMEOUT = 10000;
+  private static final int DEFAULT_REACHABILITY_THRESHOLD = 10;
+
+  private int broadcastInterval = DEFAULT_BROADCAST_INTERVAL;
+  private int reachabilityThreshold = DEFAULT_REACHABILITY_THRESHOLD;
+  private int reachabilityTimeout = DEFAULT_REACHABILITY_TIMEOUT;
+
+  /**
+   * Returns the reachability broadcast interval.
+   *
+   * @return the reachability broadcast interval
+   */
+  public int getBroadcastInterval() {
+    return broadcastInterval;
+  }
+
+  /**
+   * Sets the reachability broadcast interval.
+   *
+   * @param broadcastInterval the reachability broadcast interval
+   * @return the membership configuration
+   */
+  public MembershipConfig setBroadcastInterval(int broadcastInterval) {
+    this.broadcastInterval = broadcastInterval;
+    return this;
+  }
+
+  /**
+   * Returns the reachability failure detection threshold.
+   *
+   * @return the reachability failure detection threshold
+   */
+  public int getReachabilityThreshold() {
+    return reachabilityThreshold;
+  }
+
+  /**
+   * Sets the reachability failure detection threshold.
+   *
+   * @param reachabilityThreshold the reachability failure detection threshold
+   * @return the membership configuration
+   */
+  public MembershipConfig setReachabilityThreshold(int reachabilityThreshold) {
+    this.reachabilityThreshold = reachabilityThreshold;
+    return this;
+  }
+
+  /**
+   * Returns the reachability failure timeout.
+   *
+   * @return the reachability failure timeout
+   */
+  public int getReachabilityTimeout() {
+    return reachabilityTimeout;
+  }
+
+  /**
+   * Sets the reachability failure timeout.
+   *
+   * @param reachabilityTimeout the reachability failure timeout
+   * @return the membership configuration
+   */
+  public MembershipConfig setReachabilityTimeout(int reachabilityTimeout) {
+    this.reachabilityTimeout = reachabilityTimeout;
+    return this;
+  }
+}