Merge branch 'master' of http://git-wip-us.apache.org/repos/asf/samza
diff --git a/docs/learn/documentation/versioned/operations/monitoring.md b/docs/learn/documentation/versioned/operations/monitoring.md
index c11fbe0..de2497c 100644
--- a/docs/learn/documentation/versioned/operations/monitoring.md
+++ b/docs/learn/documentation/versioned/operations/monitoring.md
@@ -393,6 +393,7 @@
 | | disk-quota-bytes | Disk memory usage quota for key-value stores (in bytes). |
 | | executor-work-factor | The work factor of the run loop. A work factor of 1 indicates full throughput, while a work factor of less than 1 will introduce delays into the execution to approximate the requested work factor. The work factor is set by the disk space monitor in accordance with the disk quota policy. Given the latest percentage of available disk quota, this policy returns the work factor that should be applied. |
 | | physical-memory-mb | The physical memory used by the Samza container process (native + on heap) (in MBs). |
+| | physical-memory-utilization | The ratio between the physical memory used by the Samza container process (native + on heap) and the total physical memory of the Samza container. |
 | | <TaskName\>-<StoreName\>-restore-time | Time taken to restore task stores (per task store). |
 
 
diff --git a/gradle/dependency-versions.gradle b/gradle/dependency-versions.gradle
index 60e7766..e05b837 100644
--- a/gradle/dependency-versions.gradle
+++ b/gradle/dependency-versions.gradle
@@ -48,8 +48,8 @@
   scalaTestVersion = "3.0.1"
   slf4jVersion = "1.7.7"
   yarnVersion = "2.7.1"
-  zkClientVersion = "0.8"
-  zookeeperVersion = "3.4.6"
+  zkClientVersion = "0.11"
+  zookeeperVersion = "3.4.13"
   failsafeVersion = "1.1.0"
   jlineVersion = "3.8.2"
   jnaVersion = "4.5.1"
diff --git a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
index 28c3135..3a4ba17 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
@@ -617,13 +617,18 @@
       taskConfig,
       clock)
 
+    val containerMemoryMb : Int = new ClusterManagerConfig(config).getContainerMemoryMb
+
     val memoryStatisticsMonitor : SystemStatisticsMonitor = new StatisticsMonitorImpl()
     memoryStatisticsMonitor.registerListener(new SystemStatisticsMonitor.Listener {
       override def onUpdate(sample: SystemMemoryStatistics): Unit = {
         val physicalMemoryBytes : Long = sample.getPhysicalMemoryBytes
-        val physicalMemoryMb : Double = physicalMemoryBytes / (1024.0 * 1024.0)
+        val physicalMemoryMb : Float = physicalMemoryBytes / (1024.0F * 1024.0F)
+        val memoryUtilization : Float = physicalMemoryMb.toFloat / containerMemoryMb
         logger.debug("Container physical memory utilization (mb): " + physicalMemoryMb)
+        logger.debug("Container physical memory utilization: " + memoryUtilization)
         samzaContainerMetrics.physicalMemoryMb.set(physicalMemoryMb)
+        samzaContainerMetrics.physicalMemoryUtilization.set(memoryUtilization);
       }
     })
 
diff --git a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainerMetrics.scala b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainerMetrics.scala
index 73dcd20..46de166 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainerMetrics.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainerMetrics.scala
@@ -46,7 +46,8 @@
   val diskUsageBytes = newGauge("disk-usage-bytes", 0L)
   val diskQuotaBytes = newGauge("disk-quota-bytes", Long.MaxValue)
   val executorWorkFactor = newGauge("executor-work-factor", 1.0)
-  val physicalMemoryMb = newGauge[Double]("physical-memory-mb", 0.0F)
+  val physicalMemoryMb = newGauge("physical-memory-mb", 0.0F)
+  val physicalMemoryUtilization = newGauge("physical-memory-utilization", 0.0F)
   val containerThreadPoolSize = newGauge("container-thread-pool-size", 0L)
 
   val taskStoreRestorationMetrics: util.Map[TaskName, Gauge[Long]] = new util.HashMap[TaskName, Gauge[Long]]()
diff --git a/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/InMemoryKeyValueStore.java b/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/InMemoryKeyValueStore.java
index 1d5a36c..3b1256d 100644
--- a/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/InMemoryKeyValueStore.java
+++ b/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/InMemoryKeyValueStore.java
@@ -87,8 +87,6 @@
 
   @Override
   public void delete(byte[] key) {
-    // TODO Bug: SAMZA-2563: This double counts deletes for metrics, because put also counts a delete
-    metrics.deletes().inc();
     put(key, null);
   }
 
diff --git a/samza-kv-inmemory/src/test/java/org/apache/samza/storage/kv/inmemory/TestInMemoryKeyValueStore.java b/samza-kv-inmemory/src/test/java/org/apache/samza/storage/kv/inmemory/TestInMemoryKeyValueStore.java
index 48dbde6..7cdb235 100644
--- a/samza-kv-inmemory/src/test/java/org/apache/samza/storage/kv/inmemory/TestInMemoryKeyValueStore.java
+++ b/samza-kv-inmemory/src/test/java/org/apache/samza/storage/kv/inmemory/TestInMemoryKeyValueStore.java
@@ -239,11 +239,7 @@
     this.inMemoryKeyValueStore.delete(key(0));
     assertNull(this.inMemoryKeyValueStore.get(key(0)));
 
-    /*
-     * There is a bug in which deletes are double counted in metrics. This deletesCounter should only be invoked once
-     * when the bug is fixed.
-     */
-    verify(this.deletesCounter, times(2)).inc();
+    verify(this.deletesCounter, times(1)).inc();
   }
 
   @Test
@@ -251,11 +247,8 @@
     this.inMemoryKeyValueStore.delete(key(0));
 
     assertNull(this.inMemoryKeyValueStore.get(key(0)));
-    /*
-     * There is a bug in which deletes are double counted in metrics. This deletesCounter should only be invoked once
-     * when the bug is fixed.
-     */
-    verify(this.deletesCounter, times(2)).inc();
+
+    verify(this.deletesCounter, times(1)).inc();
   }
 
   @Test