KAFKA-20642: Support character ranges in Kafka shell globs

shell/src/main/java/org/apache/kafka/shell/glob/GlobComponent.java

@@ -43,11 +43,60 @@ private static boolean isRegularExpressionSpecialCharacter(char ch) {
      */
     private static boolean isGlobSpecialCharacter(char ch) {
         return switch (ch) {
-            case '*', '?', '\\', '{', '}' -> true;
+            case '*', '?', '\\', '{', '}', '[', ']' -> true;
             default -> false;
         };
     }
 
+    /**
+     * Appends one character inside a regular expression character class,
+     * escaping characters that would otherwise be interpreted specially.
+     */
+    private static void appendCharacterClassCharacter(StringBuilder output, char c) {
+        if (c == '\\' || c == '[') {
+            output.append('\\');
+        }
+        output.append(c);
+    }
+
+    /**
+     * Appends a glob character class as a regular expression character class.
+     * Returns the index immediately after the closing bracket.
+     */
+    private static int appendCharacterClass(String glob, int start, StringBuilder output) {
+        int i = start + 1;
+        if (i == glob.length()) {
+            throw new RuntimeException("Unterminated glob character class.");
+        }
+
+        output.append('[');
+        if (glob.charAt(i) == '!' || glob.charAt(i) == '^') {
+            output.append('^');
+            i++;
+        }
+        if (i == glob.length() || glob.charAt(i) == ']') {
+            throw new RuntimeException("Empty glob character class.");
+        }
+
+        for (; i < glob.length(); i++) {
+            char c = glob.charAt(i);
+            if (c == ']') {
+                output.append(']');
+                return i + 1;
+            } else if (c == '\\') {
+                if (i + 1 == glob.length()) {
+                    output.append("\\\\");
+                } else {
+                    appendCharacterClassCharacter(output, glob.charAt(++i));
+                }
+            } else {
+                appendCharacterClassCharacter(output, c);
+            }
+        }
+
+        throw new RuntimeException("Unterminated glob character class.");
+    }
+
     /**
      * Converts a glob string to a regular expression string.
      * Returns null if the glob should be handled as a literal (can only match one string).
@@ -107,7 +156,10 @@ static String toRegularExpression(String glob) {
                         output.append(c);
                     }
                     break;
-                // TODO: handle character ranges
+                case '[':
+                    literal = false;
+                    i = appendCharacterClass(glob, i - 1, output);
+                    break;
                 default:
                     if (isRegularExpressionSpecialCharacter(c)) {
                         output.append('\\');

shell/src/test/java/org/apache/kafka/shell/glob/GlobComponentTest.java

@@ -23,6 +23,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @Timeout(value = 120)
@@ -50,6 +51,17 @@ public void testToRegularExpression() {
         assertEquals("^\\$blah.*$", GlobComponent.toRegularExpression("$blah*"));
         assertEquals("^.*$", GlobComponent.toRegularExpression("*"));
         assertEquals("^foo(?:(?:bar)|(?:baz))$", GlobComponent.toRegularExpression("foo{bar,baz}"));
+        assertEquals("^topic-[0-9]$", GlobComponent.toRegularExpression("topic-[0-9]"));
+        assertEquals("^topic-[abc]$", GlobComponent.toRegularExpression("topic-[abc]"));
+        assertEquals("^topic-[^abc]$", GlobComponent.toRegularExpression("topic-[!abc]"));
+        assertEquals("^topic-[^abc]$", GlobComponent.toRegularExpression("topic-[^abc]"));
+    }
+
+    @Test
+    public void testMalformedCharacterClass() {
+        assertThrows(RuntimeException.class, () -> GlobComponent.toRegularExpression("topic-[abc"));
+        assertThrows(RuntimeException.class, () -> GlobComponent.toRegularExpression("topic-[]"));
+        assertThrows(RuntimeException.class, () -> GlobComponent.toRegularExpression("topic-[!]"));
     }
 
     @Test
@@ -71,5 +83,21 @@ public void testGlobMatch() {
         assertFalse(foobarOrFoobaz.matches("foobah"));
         assertFalse(foobarOrFoobaz.matches("foo"));
         assertFalse(foobarOrFoobaz.matches("baz"));
+        GlobComponent digit = new GlobComponent("topic-[0-9]");
+        assertFalse(digit.literal());
+        assertTrue(digit.matches("topic-0"));
+        assertTrue(digit.matches("topic-5"));
+        assertFalse(digit.matches("topic-a"));
+        assertFalse(digit.matches("topic-10"));
+        GlobComponent letters = new GlobComponent("topic-[abc]");
+        assertFalse(letters.literal());
+        assertTrue(letters.matches("topic-a"));
+        assertTrue(letters.matches("topic-b"));
+        assertFalse(letters.matches("topic-d"));
+        GlobComponent notLetters = new GlobComponent("topic-[!abc]");
+        assertFalse(notLetters.literal());
+        assertTrue(notLetters.matches("topic-d"));
+        assertTrue(notLetters.matches("topic-1"));
+        assertFalse(notLetters.matches("topic-a"));
     }
 }