Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java (Java)

Code before:
                            indent("\t\t", events.getOrDefault("max", emptyMap()).getOrDefault(key, new LinkedList<>())),
                            indent("\t\t", events.getOrDefault("min", emptyMap()).getOrDefault(key, new LinkedList<>())),
                            indent("\t\t", events.getOrDefault("dif", emptyMap()).getOrDefault(key, new LinkedList<>())),
                            indent("\t\t", events.getOrDefault("cnt", emptyMap()).getOrDefault(key, new LinkedList<>())),
                            indent("\t\t", events.getOrDefault("tagg", emptyMap()).getOrDefault(key, new LinkedList<>())));

                        if (!Set.of("echo", "max", "min", "dif", "cnt", "tagg").contains(topic))
                            resultStream.printf("%sEvents=%n%s%n", topic, indent("\t\t", entry.getValue()));
                    }

                    return false;
                }
            }
            return true;
        }
    }

    private static Boolean lessEquals(final Number expected, final Number actual) {
        if (actual instanceof Integer && expected instanceof Integer) {
            return actual.intValue() >= expected.intValue();
        } else if (actual instanceof Long && expected instanceof Long) {
            return actual.longValue() >= expected.longValue();
        } else if (actual instanceof Double && expected instanceof Double) {
            return actual.doubleValue() >= expected.doubleValue();
        } else {
            throw new RuntimeException("Unexpected type: " + actual.getClass());
        }
    }

    private static boolean verifySuppressed(final PrintStream resultStream,
                                            @SuppressWarnings("SameParameterValue") final String topic,
                                            final Map<String, Map<String, LinkedList<ConsumerRecord<String, Number>>>> events,
                                            final boolean printResults) {
        resultStream.println("verifying suppressed " + topic);
        final Map<String, LinkedList<ConsumerRecord<String, Number>>> topicEvents = events.getOrDefault(topic, emptyMap());
        for (final Map.Entry<String, LinkedList<ConsumerRecord<String, Number>>> entry : topicEvents.entrySet()) {
            if (entry.getValue().size() != 1) {
                final String unsuppressedTopic = topic.replace("-suppressed", "-raw");
                final String key = entry.getKey();
                final String unwindowedKey = key.substring(1, key.length() - 1).replaceAll("@.*", "");
                resultStream.printf("fail: key=%s%n\tnon-unique result:%n%s%n",
                                    key,
                                    indent("\t\t", entry.getValue()));

                if (printResults)
                    resultStream.printf("\tresultEvents:%n%s%n\tinputEvents:%n%s%n",
                        indent("\t\t", events.get(unsuppressedTopic).get(key)),
                        indent("\t\t", events.get("data").get(unwindowedKey)));

                return false;
            }

Review comment (suggestion):
Use a more specific exception type such as `IllegalArgumentException` for unexpected input rather than a generic `RuntimeException`.
```suggestion
            throw new IllegalArgumentException("Unexpected type: " + actual.getClass());
```

Code after:
                            indent("\t\t", events.getOrDefault("max", emptyMap()).getOrDefault(key, new LinkedList<>())),
                            indent("\t\t", events.getOrDefault("min", emptyMap()).getOrDefault(key, new LinkedList<>())),
                            indent("\t\t", events.getOrDefault("dif", emptyMap()).getOrDefault(key, new LinkedList<>())),
                            indent("\t\t", events.getOrDefault("cnt", emptyMap()).getOrDefault(key, new LinkedList<>())),
                            indent("\t\t", events.getOrDefault("tagg", emptyMap()).getOrDefault(key, new LinkedList<>())));

                        if (!Set.of("echo", "max", "min", "dif", "cnt", "tagg").contains(topic))
                            resultStream.printf("%sEvents=%n%s%n", topic, indent("\t\t", entry.getValue()));
                    }

                    return false;
                }
            }
            return true;
        }
    }

    private static boolean lessEquals(final Number expected, final Number actual) {
        if (actual instanceof Integer && expected instanceof Integer) {
            return actual.intValue() >= expected.intValue();
        } else if (actual instanceof Long && expected instanceof Long) {
            return actual.longValue() >= expected.longValue();
        } else if (actual instanceof Double && expected instanceof Double) {
            return actual.doubleValue() >= expected.doubleValue();
        } else {
            throw new IllegalArgumentException("Unexpected type: " + actual.getClass());
        }
    }

    private static boolean verifySuppressed(final PrintStream resultStream,
                                            @SuppressWarnings("SameParameterValue") final String topic,
                                            final Map<String, Map<String, LinkedList<ConsumerRecord<String, Number>>>> events,
                                            final boolean printResults) {
        resultStream.println("verifying suppressed " + topic);
        final Map<String, LinkedList<ConsumerRecord<String, Number>>> topicEvents = events.getOrDefault(topic, emptyMap());
        for (final Map.Entry<String, LinkedList<ConsumerRecord<String, Number>>> entry : topicEvents.entrySet()) {
            if (entry.getValue().size() != 1) {
                final String unsuppressedTopic = topic.replace("-suppressed", "-raw");
                final String key = entry.getKey();
                final String unwindowedKey = key.substring(1, key.length() - 1).replaceAll("@.*", "");
                resultStream.printf("fail: key=%s%n\tnon-unique result:%n%s%n",
                                    key,
                                    indent("\t\t", entry.getValue()));

                if (printResults)
                    resultStream.printf("\tresultEvents:%n%s%n\tinputEvents:%n%s%n",
                        indent("\t\t", events.get(unsuppressedTopic).get(key)),
                        indent("\t\t", events.get("data").get(unwindowedKey)));

                return false;
            }