Kafka-15126: Range queries to accept null lower and upper bounds

streams/src/main/java/org/apache/kafka/streams/query/RangeQuery.java

@@ -48,13 +48,14 @@ private RangeQuery(final Optional<K> lower, final Optional<K> upper) {
 
     /**
      * Interactive range query using a lower and upper bound to filter the keys returned.
+     * If both <K,V> are null, RangQuery returns a full range scan.
      * @param lower The key that specifies the lower bound of the range
      * @param upper The key that specifies the upper bound of the range
      * @param <K> The key type
      * @param <V> The value type
      */
     public static <K, V> RangeQuery<K, V> withRange(final K lower, final K upper) {
-        return new RangeQuery<>(Optional.of(lower), Optional.of(upper));
+        return new RangeQuery<>(Optional.ofNullable(lower), Optional.ofNullable(upper));
     }
 
     /**

streams/src/test/java/org/apache/kafka/streams/integration/IQv2StoreIntegrationTest.java

@@ -1083,22 +1083,15 @@ public <V> void shouldHandleKeyQuery(
     }   
 
     public <V> void shouldHandleRangeQuery(
+
         final Optional<Integer> lower,
         final Optional<Integer> upper,
         final Function<V, Integer> valueExtactor,
         final Set<Integer> expectedValue) {
 
         final RangeQuery<Integer, V> query;
-        if (lower.isPresent() && upper.isPresent()) {
-            query = RangeQuery.withRange(lower.get(), upper.get());
-        } else if (lower.isPresent()) {
-            query = RangeQuery.withLowerBound(lower.get());
-        } else if (upper.isPresent()) {
-            query = RangeQuery.withUpperBound(upper.get());
-        } else {
-            query = RangeQuery.withNoBounds();
-        }
 
+        query = RangeQuery.withRange(lower.orElse(null), upper.orElse(null));
         final StateQueryRequest<KeyValueIterator<Integer, V>> request =
             inStore(STORE_NAME)
                 .withQuery(query)
@@ -1376,4 +1369,4 @@ private static Properties streamsConfiguration(final boolean cache, final boolea
         config.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
         return config;
     }
-}
+}