HOTFIX: RocksDBMetricsRecorder#init should null check taskId (apache#…

…18151)

Appears to be a typo in the code, since the error message indicates this check is for taskId being null, but instead we accidentally check the streams metrics twice

Reviewers: Matthias Sax <[email protected]>, runo Cadonna <[email protected]>, Lucas Brutschy <[email protected]>, Bill Bejeck <[email protected]>

streams/src/main/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorder.java

@@ -135,7 +135,7 @@ public TaskId taskId() {
     public void init(final StreamsMetricsImpl streamsMetrics,
                      final TaskId taskId) {
         Objects.requireNonNull(streamsMetrics, "Streams metrics must not be null");
-        Objects.requireNonNull(streamsMetrics, "task ID must not be null");
+        Objects.requireNonNull(taskId, "task ID must not be null");
         if (this.taskId != null && !this.taskId.equals(taskId)) {
             throw new IllegalStateException("Metrics recorder is re-initialised with different task: previous task is " +
                 this.taskId + " whereas current task is " + taskId + ". This is a bug in Kafka Streams. " +

streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderTest.java

@@ -173,6 +173,22 @@ public void shouldThrowIfMetricRecorderIsReInitialisedWithDifferentTask() {
         );
     }
 
+    @Test
+    public void shouldThrowIfMetricRecorderIsInitialisedWithNullMetrics() {
+        assertThrows(
+            NullPointerException.class,
+            () -> recorder.init(null, TASK_ID1)
+        );
+    }
+
+    @Test
+    public void shouldThrowIfMetricRecorderIsInitialisedWithNullTaskId() {
+        assertThrows(
+            NullPointerException.class,
+            () -> recorder.init(streamsMetrics, null)
+        );
+    }
+
     @Test
     public void shouldThrowIfMetricRecorderIsReInitialisedWithDifferentStreamsMetrics() {
         assertThrows(