[kafka] branch trunk updated: MINOR: No error with zero results state query (#13002)

[bb...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

bbejeck pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new ea65d74f6bc MINOR: No error with zero results state query (#13002)
ea65d74f6bc is described below

commit ea65d74f6bc7f4a94f7f037cbf0103e925a9ddbb
Author: Bill Bejeck <bb...@gmail.com>
AuthorDate: Mon Dec 19 13:39:06 2022 -0500

    MINOR: No error with zero results state query (#13002)
    
    This PR updates StateQueryResult.getOnlyPartitionResult() to not throw an IllegaArgumentException when there are 0 query results.
    
    Added a test that will fail without this patch
    
    Reviewers: John Roesler<vv...@apache.org>
---
 .../kafka/streams/query/StateQueryResult.java      |  4 +-
 .../kafka/streams/query/StateQueryResultTest.java  | 64 ++++++++++++++++++++++
 2 files changed, 66 insertions(+), 2 deletions(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/query/StateQueryResult.java b/streams/src/main/java/org/apache/kafka/streams/query/StateQueryResult.java
index c5eaa10663a..722a1851660 100644
--- a/streams/src/main/java/org/apache/kafka/streams/query/StateQueryResult.java
+++ b/streams/src/main/java/org/apache/kafka/streams/query/StateQueryResult.java
@@ -72,12 +72,12 @@ public class StateQueryResult<R> {
                 .filter(r -> r.getResult() != null)
                 .collect(Collectors.toList());
 
-        if (nonempty.size() != 1) {
+        if (nonempty.size() > 1) {
             throw new IllegalArgumentException(
                 "The query did not return exactly one partition result: " + partitionResults
             );
         } else {
-            return nonempty.get(0);
+            return nonempty.isEmpty() ? null : nonempty.get(0);
         }
     }
 
diff --git a/streams/src/test/java/org/apache/kafka/streams/query/StateQueryResultTest.java b/streams/src/test/java/org/apache/kafka/streams/query/StateQueryResultTest.java
new file mode 100644
index 00000000000..0e9313da8d3
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/query/StateQueryResultTest.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.query;
+
+import org.apache.kafka.streams.query.internals.SucceededQueryResult;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertThrows;
+
+
+class StateQueryResultTest {
+
+    StateQueryResult<String> stringStateQueryResult;
+    final QueryResult<String> noResultsFound = new SucceededQueryResult<>(null);
+    final QueryResult<String> validResult = new SucceededQueryResult<>("Foo");
+
+    @BeforeEach
+    public void setUp() {
+        stringStateQueryResult = new StateQueryResult<>();
+    }
+
+    @Test
+    @DisplayName("Zero query results shouldn't error")
+    void getOnlyPartitionResultNoResultsTest() {
+        stringStateQueryResult.addResult(0, noResultsFound);
+        final QueryResult<String> result = stringStateQueryResult.getOnlyPartitionResult();
+        assertThat(result, nullValue());
+    }
+
+    @Test
+    @DisplayName("Valid query results still works")
+    void getOnlyPartitionResultWithSingleResultTest() {
+        stringStateQueryResult.addResult(0, validResult);
+        final QueryResult<String> result = stringStateQueryResult.getOnlyPartitionResult();
+        assertThat(result.getResult(), is("Foo"));
+    }
+
+    @Test
+    @DisplayName("More than one query result throws IllegalArgumentException ")
+    void getOnlyPartitionResultMultipleResults() {
+        stringStateQueryResult.addResult(0, validResult);
+        stringStateQueryResult.addResult(1, validResult);
+        assertThrows(IllegalArgumentException.class, () -> stringStateQueryResult.getOnlyPartitionResult());
+    }
+}
\ No newline at end of file