[HOT FIX] Check for null before deserializing in MeteredSessionStore

streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStore.java

@@ -151,30 +151,41 @@ public void remove(final Windowed<K> sessionKey) {
     @Override
     public V fetchSession(final K key, final long startTime, final long endTime) {
         Objects.requireNonNull(key, "key cannot be null");
-        final V value;
         final Bytes bytesKey = keyBytes(key);
         final long startNs = time.nanoseconds();
         try {
-            value = serdes.valueFrom(wrapped().fetchSession(bytesKey, startTime, endTime));
+            final byte[] result = wrapped().fetchSession(bytesKey, startTime, endTime);
+            if (result == null) {
+                return null;
+            }
+            return serdes.valueFrom(result);
         } finally {
             metrics.recordLatency(flushTime, startNs, time.nanoseconds());
         }
-
-        return value;
     }
 
     @Override
     public KeyValueIterator<Windowed<K>, V> fetch(final K key) {
         Objects.requireNonNull(key, "key cannot be null");
-        return findSessions(key, 0, Long.MAX_VALUE);
+        return new MeteredWindowedKeyValueIterator<>(
+            wrapped().fetch(keyBytes(key)),
+            fetchTime,
+            metrics,
+            serdes,
+            time);
     }
 
     @Override
     public KeyValueIterator<Windowed<K>, V> fetch(final K from,
                                                   final K to) {
         Objects.requireNonNull(from, "from cannot be null");
         Objects.requireNonNull(to, "to cannot be null");
-        return findSessions(from, to, 0, Long.MAX_VALUE);
+        return new MeteredWindowedKeyValueIterator<>(
+            wrapped().fetch(keyBytes(from), keyBytes(to)),
+            fetchTime,
+            metrics,
+            serdes,
+            time);
     }
 
     @Override

streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStoreTest.java

@@ -55,6 +55,7 @@
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 @RunWith(EasyMockRunner.class)
@@ -243,6 +244,14 @@ public void shouldSetFlushListenerOnWrappedCachingStore() {
         verify(cachedKeyValueStore);
     }
 
+    @Test
+    public void shouldNotThrowNullPointerExceptionIfGetReturnsNull() {
+        expect(inner.get(Bytes.wrap("a".getBytes()))).andReturn(null);
+
+        init();
+        assertNull(metered.get("a"));
+    }
+
     @Test
     public void shouldNotSetFlushListenerOnWrappedNoneCachingStore() {
         assertFalse(metered.setFlushListener(null, false));

streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreTest.java

@@ -57,6 +57,7 @@
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 @RunWith(EasyMockRunner.class)
@@ -172,7 +173,7 @@ public void shouldRemoveFromStoreAndRecordRemoveMetric() {
 
     @Test
     public void shouldFetchForKeyAndRecordFetchMetric() {
-        expect(inner.findSessions(Bytes.wrap(keyBytes), 0, Long.MAX_VALUE))
+        expect(inner.fetch(Bytes.wrap(keyBytes)))
                 .andReturn(new KeyValueIteratorStub<>(
                         Collections.singleton(KeyValue.pair(windowedKeyBytes, keyBytes)).iterator()));
         init();
@@ -189,7 +190,7 @@ public void shouldFetchForKeyAndRecordFetchMetric() {
 
     @Test
     public void shouldFetchRangeFromStoreAndRecordFetchMetric() {
-        expect(inner.findSessions(Bytes.wrap(keyBytes), Bytes.wrap(keyBytes), 0, Long.MAX_VALUE))
+        expect(inner.fetch(Bytes.wrap(keyBytes), Bytes.wrap(keyBytes)))
                 .andReturn(new KeyValueIteratorStub<>(
                         Collections.singleton(KeyValue.pair(windowedKeyBytes, keyBytes)).iterator()));
         init();
@@ -211,6 +212,14 @@ public void shouldRecordRestoreTimeOnInit() {
         assertTrue((Double) metric.metricValue() > 0);
     }
 
+    @Test
+    public void shouldNotThrowNullPointerExceptionIfFetchSessionReturnsNull() {
+        expect(inner.fetchSession(Bytes.wrap("a".getBytes()), 0, Long.MAX_VALUE)).andReturn(null);
+
+        init();
+        assertNull(metered.fetchSession("a", 0, Long.MAX_VALUE));
+    }
+
     @Test(expected = NullPointerException.class)
     public void shouldThrowNullPointerOnPutIfKeyIsNull() {
         metered.put(null, "a");

streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredWindowStoreTest.java

@@ -178,7 +178,7 @@ public void shouldCloseUnderlyingStore() {
     }
 
     @Test
-    public void shouldNotExceptionIfFetchReturnsNull() {
+    public void shouldNotThrowNullPointerExceptionIfFetchReturnsNull() {
         expect(innerStoreMock.fetch(Bytes.wrap("a".getBytes()), 0)).andReturn(null);
         replay(innerStoreMock);