Repository: incubator-samza
Updated Branches:
  refs/heads/master 74b56d639 -> f5741125a


SAMZA-50: Log both RMI and JMX ports in internal JMX server.


Project: http://git-wip-us.apache.org/repos/asf/incubator-samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-samza/commit/f5741125
Tree: http://git-wip-us.apache.org/repos/asf/incubator-samza/tree/f5741125
Diff: http://git-wip-us.apache.org/repos/asf/incubator-samza/diff/f5741125

Branch: refs/heads/master
Commit: f5741125a17f7875054f8c2d9419743d3da111d5
Parents: 74b56d6
Author: Jakob Glen Homan <[email protected]>
Authored: Tue Mar 4 16:24:11 2014 -0800
Committer: Jakob Glen Homan <[email protected]>
Committed: Tue Mar 4 16:24:11 2014 -0800

----------------------------------------------------------------------
 .../org/apache/samza/metrics/JmxServer.scala    | 26 +++++++++++++-------
 .../apache/samza/metrics/TestJmxServer.scala    |  3 ++-
 2 files changed, 19 insertions(+), 10 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/f5741125/samza-core/src/main/scala/org/apache/samza/metrics/JmxServer.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/metrics/JmxServer.scala 
b/samza-core/src/main/scala/org/apache/samza/metrics/JmxServer.scala
index eee213d..0ff88e7 100644
--- a/samza-core/src/main/scala/org/apache/samza/metrics/JmxServer.scala
+++ b/samza-core/src/main/scala/org/apache/samza/metrics/JmxServer.scala
@@ -39,10 +39,12 @@ import org.apache.samza.config.Config
  * @param requestedPort Port on which to start JMX server, 0 for ephemeral
  */
 class JmxServer(requestedPort: Int) extends Logging {
+  val hostname = InetAddress.getLocalHost.getHostName
+
   def this() = this(0)
 
   // Instance construction
-  val (jmxServer, url, actualPort) = {
+  val (jmxServer, url, registryPort, serverPort) = {
     // An RMIServerSocketFactory that will tell what port it opened up.  
Imagine that.
     class UpfrontRMIServerSocketFactory extends RMIServerSocketFactory {
       var lastSS: ServerSocket = null
@@ -68,7 +70,6 @@ class JmxServer(requestedPort: Int) extends Logging {
         "This behavior is not well defined and our values will collide with 
any set on command line.")
     }
 
-    val hostname = InetAddress.getLocalHost.getHostName
     info("According to InetAddress.getLocalHost.getHostName we are " + 
hostname)
     updateSystemProperty("com.sun.management.jmxremote.authenticate", "false")
     updateSystemProperty("com.sun.management.jmxremote.ssl", "false")
@@ -76,23 +77,31 @@ class JmxServer(requestedPort: Int) extends Logging {
 
     val ssFactory = new UpfrontRMIServerSocketFactory
     LocateRegistry.createRegistry(requestedPort, null, ssFactory)
-    val actualPort = ssFactory.lastSS.getLocalPort
+    val registryPort = ssFactory.lastSS.getLocalPort
+    val serverPort = registryPort + 1 // In comparison to the registry port. 
Tiny chance of collision.  Sigh.
     val mbs = ManagementFactory.getPlatformMBeanServer
     val env = new util.HashMap[String, Object]()
-    val url = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://" + hostname + 
":" + actualPort + "/jmxrmi")
+    val url = new JMXServiceURL("service:jmx:rmi://localhost:" + serverPort + 
"/jndi/rmi://localhost:" + registryPort + "/jmxrmi")
     val jmxServer = JMXConnectorServerFactory.newJMXConnectorServer(url, env, 
mbs)
 
-    (jmxServer, url.toString, actualPort)
+    (jmxServer, url.toString, registryPort, serverPort)
   }
 
   jmxServer.start
   info("Started " + toString)
+  info("If you are tunneling, you might want to try " + 
toString.replaceAll("localhost", hostname))
 
   /**
-   * Get RMI port the JMX server is listening on.
+   * Get RMI registry port
    * @return RMI port
    */
-  def getPort = actualPort
+  def getRegistryPort = registryPort
+
+  /**
+   * Get JMX server port
+   */
+  def getServerPort = serverPort
+
 
   /**
    * Get Jmx URL for this server
@@ -105,6 +114,5 @@ class JmxServer(requestedPort: Int) extends Logging {
    */
   def stop = jmxServer.stop
 
-  override def toString = "JmxServer port=%d url=%s" format (getPort, 
getJmxUrl)
+  override def toString = "JmxServer registry port=%d server port=%d url=%s" 
format (getRegistryPort, getServerPort, getJmxUrl)
 }
-

http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/f5741125/samza-core/src/test/scala/org/apache/samza/metrics/TestJmxServer.scala
----------------------------------------------------------------------
diff --git 
a/samza-core/src/test/scala/org/apache/samza/metrics/TestJmxServer.scala 
b/samza-core/src/test/scala/org/apache/samza/metrics/TestJmxServer.scala
index f5594d0..4f7ddcd 100644
--- a/samza-core/src/test/scala/org/apache/samza/metrics/TestJmxServer.scala
+++ b/samza-core/src/test/scala/org/apache/samza/metrics/TestJmxServer.scala
@@ -35,7 +35,8 @@ class TestJmxServer extends Logging {
     try {
       jmxServer = new JmxServer
 
-      println("Got jmxServer on port " + jmxServer.getPort)
+      println("JmxServer = %s" format jmxServer)
+      println("Got jmxServer on port " + jmxServer.getRegistryPort)
 
       val jmxURL = new JMXServiceURL(jmxServer.getJmxUrl)
       var jmxConnector:JMXConnector = null

Reply via email to