add units to metrics descriptions + test fix

clients/src/main/java/org/apache/kafka/clients/producer/internals/KafkaProducerMetrics.java

@@ -61,15 +61,15 @@ public KafkaProducerMetrics(Metrics metrics) {
         );
         sendOffsetsSensor = newLatencySensor(
             TXN_SEND_OFFSETS,
-            "Total time producer has spent in sendOffsetsToTransaction."
+            "Total time producer has spent in sendOffsetsToTransaction in nanoseconds."
         );
         commitTxnSensor = newLatencySensor(
             TXN_COMMIT,
-            "Total time producer has spent in commitTransaction."
+            "Total time producer has spent in commitTransaction in nanoseconds."
         );
         abortTxnSensor = newLatencySensor(
             TXN_ABORT,
-            "Total time producer has spent in abortTransaction."
+            "Total time producer has spent in abortTransaction in nanoseconds."
         );
     }
 

clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java

@@ -1061,7 +1061,7 @@ private void assertDurationAtLeast(KafkaProducer<?, ?> producer, String name, do
 
     private double getAndAssertDurationAtLeast(KafkaProducer<?, ?> producer, String name, double floor) {
         double value = getMetricValue(producer, name);
-        assertTrue(value > floor);
+        assertTrue(value >= floor);
         return value;
     }
 

streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/ThreadMetrics.java

@@ -95,7 +95,7 @@ private ThreadMetrics() {}
     private static final String COMMIT_RATIO_DESCRIPTION =
         "The fraction of time the thread spent on committing all tasks";
     private static final String BLOCKED_TIME_DESCRIPTION =
-        "The total time the thread spent blocked on kafka";
+        "The total time the thread spent blocked on kafka in nanoseconds";
     private static final String THREAD_START_TIME_DESCRIPTION =
         "The time that the thread was started";
 

streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/ThreadMetricsTest.java

@@ -433,7 +433,7 @@ public void shouldAddTotalBlockedTimeMetric() {
         final ArgumentCaptor<Gauge<Double>> captor = gaugeCaptor();
         verify(streamsMetrics).addThreadLevelMutableMetric(
             eq("blocked-time-ns-total"),
-            eq("The total time the thread spent blocked on kafka"),
+            eq("The total time the thread spent blocked on kafka in nanoseconds"),
             eq("burger"),
             captor.capture()
         );