From 741ff8fdbd66844209d5883fd6509219dc2823b1 Mon Sep 17 00:00:00 2001
From: Piotr Findeisen <piotr.findeisen@gmail.com>
Date: Mon, 22 Mar 2021 09:24:01 +0100
Subject: [PATCH] Check for subject being just -value / -key

Such topic is probably illegal, but let's make the code immune to it
anyway.
---
 ...entSchemaRegistryTableDescriptionSupplier.java | 15 +++++++++++----
 1 file changed, 11 insertions(+), 4 deletions(-)

diff --git a/plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/schema/confluent/ConfluentSchemaRegistryTableDescriptionSupplier.java b/plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/schema/confluent/ConfluentSchemaRegistryTableDescriptionSupplier.java
index 3ce604eb52460..2bcc75e46d754 100644
--- a/plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/schema/confluent/ConfluentSchemaRegistryTableDescriptionSupplier.java
+++ b/plugin/trino-kafka/src/main/java/io/trino/plugin/kafka/schema/confluent/ConfluentSchemaRegistryTableDescriptionSupplier.java
@@ -41,6 +41,7 @@
 import java.util.Set;
 import java.util.function.Supplier;
 
+import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.base.Suppliers.memoizeWithExpiration;
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
@@ -272,17 +273,23 @@ public Set<SchemaTableName> listTables()
     private static boolean isValidSubject(String subject)
     {
         requireNonNull(subject, "subject is null");
-        return subject.endsWith(VALUE_SUFFIX) || subject.endsWith(KEY_SUFFIX);
+        return (subject.endsWith(VALUE_SUFFIX) && subject.length() > VALUE_SUFFIX.length())
+                || (subject.endsWith(KEY_SUFFIX) && subject.length() > KEY_SUFFIX.length());
     }
 
     private static String extractTopicFromSubject(String subject)
     {
         requireNonNull(subject, "subject is null");
+        String topic;
         if (subject.endsWith(VALUE_SUFFIX)) {
-            return subject.substring(0, subject.length() - VALUE_SUFFIX.length());
+            topic = subject.substring(0, subject.length() - VALUE_SUFFIX.length());
         }
-        checkState(subject.endsWith(KEY_SUFFIX), "Unexpected subject name %s", subject);
-        return subject.substring(0, subject.length() - KEY_SUFFIX.length());
+        else {
+            checkState(subject.endsWith(KEY_SUFFIX), "Unexpected subject name %s", subject);
+            topic = subject.substring(0, subject.length() - KEY_SUFFIX.length());
+        }
+        checkArgument(!topic.isEmpty(), "Unexpected subject name %s", subject);
+        return topic;
     }
 
     private static Optional<String> getKeySubjectFromTopic(String topic, Collection<String> subjectsForTopic)