Fixing a bug in the way the JMX bean names are picked

.gitignore

@@ -1,5 +1,14 @@
+bin/
 target/
 lib_managed/
 src_managed/
 project/boot/
 project/plugins/project
+out/
+build/
+.idea/
+.gradle
+*.iml
+*.ipr
+*.iws
+*.swp

build.gradle

@@ -0,0 +1,48 @@
+project.ext.isDefaultEnvironment = !project.hasProperty('overrideBuildEnvironment')
+
+File getEnvironmentScript()
+{
+  final File env = file(isDefaultEnvironment ? 'defaultEnvironment.gradle' : project.overrideBuildEnvironment)
+  assert env.isFile() : "The environment script [$env] does not exists or is not a file."
+  return env
+}
+
+apply from: environmentScript
+
+def getScalaSuffix(scalaVersion) {
+  return scalaVersion.startsWith('2.10') ? '_2.10' : "_$scalaVersion"
+}
+
+subprojects {
+
+  // the cross built scala modules share the same source directories so we need to make their output directories unique
+  buildDir = "${rootProject.buildDir}/$name"
+
+  plugins.withType(JavaPlugin) {
+    project.dependencies {
+      testCompile externalDependency.junit
+    }
+  }
+
+  def projectScalaVersion = properties.targetScalaVersions.split(',').find { name.contains(getScalaSuffix(it)) }
+  ext.scalaVersion = projectScalaVersion ? projectScalaVersion : properties.defaultScalaVersion
+  ext.scalaSuffix = getScalaSuffix(ext.scalaVersion)
+
+  ext.externalDependency = [
+    'zookeeper':'org.apache.zookeeper:zookeeper:3.3.4',
+    'protobuf':'com.google.protobuf:protobuf-java:2.4.0a',
+    'log4j':'log4j:log4j:1.2.17',
+    'netty':'io.netty:netty:3.7.0.Final',
+    'slf4jApi':'org.slf4j:slf4j-api:1.7.5',
+    'slf4jLog4j':'org.slf4j:slf4j-log4j12:1.7.5',
+    'specs':"org.scala-tools.testing:specs${ext.scalaSuffix}:${ext.scalaSuffix == '_2.10' ? '1.6.9' : '1.6.8'}",
+    'mockitoAll':'org.mockito:mockito-all:1.8.4',
+    'cglib':'cglib:cglib:2.1_3',
+    'objenesis':'org.objenesis:objenesis:1.2',
+    'scalaCompiler': "org.scala-lang:scala-compiler:${ext.scalaVersion}",
+    'scalaLibrary': "org.scala-lang:scala-library:${ext.scalaVersion}",
+    'scalaActors': "org.scala-lang:scala-actors:${ext.scalaVersion}",
+    'junit':'junit:junit:4.8.1'
+  ];
+
+}

cluster/build.gradle

@@ -0,0 +1,19 @@
+apply plugin: 'java'
+apply plugin: 'scala'
+
+dependencies {
+  compile externalDependency.scalaLibrary
+  compile externalDependency.zookeeper
+  compile externalDependency.protobuf
+  compile externalDependency.log4j
+  if (scalaVersion.startsWith('2.10')) {
+    compile externalDependency.scalaActors
+  }
+
+  testCompile externalDependency.specs
+  testCompile externalDependency.mockitoAll
+  testCompile externalDependency.cglib
+  testCompile externalDependency.objenesis
+  testCompile externalDependency.junit
+}
+

cluster/src/main/scala/com/linkedin/norbert/cluster/ClusterClient.scala

@@ -24,6 +24,7 @@ import jmx.JMX.MBean
 import jmx.JMX
 import logging.Logging
 import zookeeper.ZooKeeperClusterClient
+import java.net.InetAddress
 
 /**
  * ClusterClient companion object provides factory methods for creating a <code>ClusterClient</code> instance.
@@ -115,6 +116,21 @@ trait ClusterClient extends Logging {
     }
   }
 
+  /**
+   * Returns the next node id that isn't taken
+   *
+   * @return the current list of nodes
+   * @throws ClusterDisconnectedException thrown if the cluster is not connected when the method is called
+   */
+  def nextNodeId: Int = {
+    val ns = nodes
+    if (ns.size == 0) {
+      0
+    } else {
+      ns.map(_.id).max + 1
+    }
+  }
+
   /**
    * Looks up the node with the specified id.
    *
@@ -125,6 +141,68 @@ trait ClusterClient extends Logging {
    */
   def nodeWithId(nodeId: Int): Option[Node] = nodeWith(_.id == nodeId)
 
+  /**
+   * Looks up the node with with the specified port for the machine which is running this code.
+   *
+   * @param port the port of the current hostname of the node to find
+   *
+   * @return <code>Some</code> with the node if found, otherwise <code>None</code>
+   * @throws ClusterDisconnectedException thrown if the cluster is not connected when the method is called
+   */
+  def myNodeByPort(port: Int): Option[Node] = {
+    val url = InetAddress.getLocalHost.getCanonicalHostName
+    val node = nodeByUrl(url, port)
+    if(node == None) {
+      log.warn("No node could be found for this machine's URL and port: %s:%d".format(url, port))
+    }
+    node
+  }
+
+  /**
+   * Looks up the node with with the specified url and port.
+   *
+   * @param hostname the hostname of the node to find
+   * @param port the id of the node to find
+   *
+   * @return <code>Some</code> with the node if found, otherwise <code>None</code>
+   * @throws ClusterDisconnectedException thrown if the cluster is not connected when the method is called
+   */
+  def nodeByUrl(hostname:String, port:Int) = nodeWith(_.url == hostname + ":" + port)
+
+  /**
+   * Adds a node to the cluster metadata, picks an id above all other ids
+   *
+   * @param url the url to be used to send requests to the node
+   *
+   * @return the newly added node
+   * @throws ClusterDisconnectedException thrown if the cluster is disconnected when the method is called
+   * @throws InvalidNodeException thrown if there is an error adding the new node to the cluster metadata
+   */
+   def addNodeByUrl(hostname: String, port:Int, partitionBuilder:Option[Int => Set[Int]] = None): Node = {
+    val url = hostname + ":" + port
+    var id =  nextNodeId
+    var node: Node = null
+    val pbuilder = partitionBuilder match {
+      case Some(f) => f
+      case None => {_:Int => Set.empty[Int]}
+    }
+    while (node == null) {
+      try {
+        node = addNode(id, url, pbuilder(id))
+      } catch {
+        case _:InvalidNodeException =>
+      }
+      val nid = nextNodeId
+      if (nid == id) {
+        /* there was a problem with the id, try one last time and return that error */
+        node = addNode(id, url, pbuilder(id))
+      }
+      id = nid
+    }
+    node
+  }
+
+
   /**
    * Adds a node to the cluster metadata.
    *

cluster/src/main/scala/com/linkedin/norbert/cluster/zookeeper/ZooKeeperClusterManagerComponent.scala

@@ -37,6 +37,8 @@ trait ZooKeeperClusterManagerComponent extends ClusterManagerComponent {
     //synchronization is not needed between this method
     case class NodeChildrenChanged(path: String) extends ZooKeeperMessage
     case class NodeDataChanged(path: String) extends ZooKeeperMessage
+    case class NodeDeleted(path: String) extends ZooKeeperMessage
+    case class NodeCreated(path: String) extends ZooKeeperMessage
   }
 
   class ZooKeeperClusterManager(connectString: String, sessionTimeout: Int, serviceName: String)
@@ -90,6 +92,19 @@ trait ZooKeeperClusterManagerComponent extends ClusterManagerComponent {
             handleCapabilityMemberChanged(path)
           }
 
+          case NodeDeleted(path) => if (path.equals(MEMBERSHIP_NODE) || path.equals(AVAILABILITY_NODE)) {
+            //zookeeper data corrupted
+            log.fatal("Received a zookeeper corruption message")    
+          } else if (path.startsWith(MEMBERSHIP_NODE) || path.startsWith(AVAILABILITY_NODE)) {
+            log.info("Received an event where a node was deleted:%s".format(path))
+          } else {
+            log.error("Node deleted unexpectedly %s".format(path))
+          }
+
+          case NodeCreated(path) => {
+            log.error("Received an unexpected create event:%s".format(path))
+          }
+
           case m => log.error("Received unknown message: %s".format(m))
         }
       }
@@ -461,6 +476,10 @@ trait ZooKeeperClusterManagerComponent extends ClusterManagerComponent {
         case EventType.NodeChildrenChanged => zooKeeperManager ! NodeChildrenChanged(event.getPath)
 
         case EventType.NodeDataChanged => zooKeeperManager ! NodeDataChanged(event.getPath)
+
+        case EventType.NodeCreated => zooKeeperManager ! NodeCreated(event.getPath)
+
+        case EventType.NodeDeleted => zooKeeperManager ! NodeDeleted(event.getPath)
       }
     }