fix: query id for TERMINATE should be case insensitive

ksqldb-common/src/main/java/io/confluent/ksql/query/QueryId.java

@@ -22,21 +22,32 @@
 import com.google.errorprone.annotations.Immutable;
 import java.util.Objects;
 
+/**
+ * A query id.
+ *
+ * <p>For backwards compatibility reasons query ids must preserve their case, as their text
+ * representation is used, amoung other things, for internal topic naming.
+ *
+ * <p>However, two ids with the same text, with different case, should compare equal. This is needed
+ * so that look ups against query ids are not case-sensitive.
+ */
 @Immutable
 public class QueryId {
 
   private final String id;
+  private final String cachedUpperCase;
 
   @JsonCreator
   public QueryId(final String id) {
     this.id = requireNonNull(id, "id");
+    this.cachedUpperCase = id.toUpperCase();
   }
 
-  @JsonValue
   public String getId() {
     return id;
   }
 
+  @JsonValue
   public String toString() {
     return id;
   }
@@ -50,11 +61,11 @@ public boolean equals(final Object o) {
       return false;
     }
     final QueryId queryId = (QueryId) o;
-    return Objects.equals(id, queryId.id);
+    return Objects.equals(cachedUpperCase, queryId.cachedUpperCase);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(id);
+    return Objects.hash(cachedUpperCase);
   }
 }

ksqldb-common/src/test/java/io/confluent/ksql/query/QueryIdTest.java

@@ -19,33 +19,61 @@
 import static org.hamcrest.Matchers.is;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.testing.EqualsTester;
 import io.confluent.ksql.json.JsonMapper;
 import org.junit.Test;
 
 public class QueryIdTest {
 
   private final ObjectMapper objectMapper = JsonMapper.INSTANCE.mapper;
 
+  @SuppressWarnings("UnstableApiUsage")
   @Test
-  public void shouldSerializeCorrectly() throws Exception {
+  public void shouldImplementEqualsProperly() {
+    new EqualsTester()
+        .addEqualityGroup(new QueryId("matching"), new QueryId("MaTcHiNg"))
+        .addEqualityGroup(new QueryId("different"))
+        .testEquals();
+  }
+
+  @Test
+  public void shouldBeCaseInsensitiveOnCommparison() {
+    // When:
+    final QueryId id = new QueryId("Mixed-Case-Id");
+
+    // Then:
+    assertThat(id, is(new QueryId("MIXED-CASE-ID")));
+  }
+
+  @Test
+  public void shouldPreserveCase() {
+    // When:
+    final QueryId id = new QueryId("Mixed-Case-Id");
+
+    // Then:
+    assertThat(id.toString(), is("Mixed-Case-Id"));
+  }
+
+  @Test
+  public void shouldSerializeMaintainingCase() throws Exception {
     // Given:
-    final QueryId id = new QueryId("query-id");
+    final QueryId id = new QueryId("Query-Id");
 
     // When:
     final String serialized = objectMapper.writeValueAsString(id);
 
-    assertThat(serialized, is("\"query-id\""));
+    assertThat(serialized, is("\"Query-Id\""));
   }
 
   @Test
-  public void shouldDeserializeCorrectly() throws Exception {
+  public void shouldDeserializeMaintainingCase() throws Exception {
     // Given:
-    final String serialized = "\"an-id\"";
+    final String serialized = "\"An-Id\"";
 
     // When:
     final QueryId deserialized = objectMapper.readValue(serialized, QueryId.class);
 
     // Then:
-    assertThat(deserialized, is(new QueryId("an-id")));
+    assertThat(deserialized.getId(), is("An-Id"));
   }
 }

ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/computation/RecoveryTest.java

@@ -568,7 +568,7 @@ public void shouldRecoverRecreates() {
     server1.submitCommands(
         "CREATE STREAM A (C1 STRING, C2 INT) WITH (KAFKA_TOPIC='A', VALUE_FORMAT='JSON');",
         "CREATE STREAM B AS SELECT C1 FROM A;",
-        "TERMINATE CSAS_B_0;",
+        "TERMINATE CsAs_b_0;",
         "DROP STREAM B;",
         "CREATE STREAM B AS SELECT C2 FROM A;"
     );