feat: add ASSERT TOPIC command #9066

jzaralim · 2022-04-26T22:32:24Z

Description

Adds ASSERT TOPIC command to ksqlDB. See discussion here

Testing done

Unit + integration tests

Reviewer checklist

Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
Ensure relevant issues are linked (description should include text like "Fixes #")

jnh5y · 2022-04-27T12:43:34Z

ksqldb-parser/src/main/antlr4/io/confluent/ksql/parser/SqlBase.g4

@@ -89,6 +89,7 @@ statement
    | CREATE TYPE (IF NOT EXISTS)? identifier AS type                       #registerType
    | DROP TYPE (IF EXISTS)? identifier                                     #dropType
    | ALTER (STREAM | TABLE) sourceName alterOption (',' alterOption)*      #alterSource
+    | ASSERT TOPIC identifier (WITH tableProperties)? timeout?              #assertTopic


Just to ask... ASSERT TOPIC needs to be here to be processed with other DDL commands instead of with assertStatement below? (That's my guess; just validating.)

Yup, the ones below are only used by YATT.

Are you adding the "NOT" support in this PR or a separate one?

It's going in a separate one.

jnh5y · 2022-04-27T12:45:54Z

ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/integration/RestApiTest.java

+      }
+    }, containsString(
+        "Mismatched configuration for topic X: For config partitions, expected apples got 1\n"
+            + "Mismatched configuration for topic X: For config replicas, expected 100 got 1\n"


Is the expected text backwards? The stream was created with (1,1) and (apples, 100) were offered, right?

its (the assert statement) expected 100, got 1 (from the topic instead)

i can see how the wording is ambiguous. maybe something like would be clearer:

For config partitions, 100 was asserted, but the actual value was 1

Ah, I read it backwards yesterday!

jnh5y · 2022-04-28T15:48:33Z

ksqldb-cli/src/test/java/io/confluent/ksql/cli/console/ConsoleTest.java

+  public void shouldPrintAssertTopicResult() {
+    // Given:
+    final KsqlEntityList entities = new KsqlEntityList(ImmutableList.of(
+        new AssertTopicEntity("statement", "topic")


As an optional thing, it may be worth considering a new other than "topic" for the topic. That avoids the approved version being "Topic topic exists."

shrugs

lol okay. Done.

jnh5y · 2022-04-28T15:49:19Z

ksqldb-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java

@@ -661,6 +661,12 @@ public class KsqlConfig extends AbstractConfig {
  private static final boolean KSQL_ENDPOINT_MIGRATE_QUERY_DEFAULT = true;
  private static final String KSQL_ENDPOINT_MIGRATE_QUERY_DOC
      = "Migrates the /query endpoint to use the same handler as /query-stream.";
+  public static final String KSQL_ASSERT_TOPIC_DEFAULT_TIMEOUT_MS


Do we need/want to document this setting?

I plan on doing docs all in one go. It saves us the trouble of having to revert things in case the full project doesn't land in the next release.

jnh5y · 2022-04-28T15:53:34Z

ksqldb-parser/src/main/java/io/confluent/ksql/parser/tree/AssertTopic.java

+      return false;
+    }
+    final AssertTopic that = (AssertTopic) o;
+    return topic.equals(that.topic)


Do we need to check that the location is the same?

Nope, location just refers to what line/position the ASSERT token was found, so it doesn't really matter here.

jnh5y · 2022-04-28T15:55:19Z

ksqldb-parser/src/test/java/io/confluent/ksql/util/GrammarTokenExporterTest.java

@@ -41,7 +41,7 @@ public class GrammarTokenExporterTest {
      "CATALOG", "PROPERTIES", "BEGINNING", "UNSET", "RUN", "SCRIPT", "DECIMAL",
      "KEY", "CONNECTOR", "CONNECTORS", "SINK", "SOURCE", "NAMESPACE", "MATERIALIZED",
      "VIEW", "PRIMARY", "REPLACE", "ASSERT", "ADD", "ALTER", "VARIABLES", "PLUGINS", "HEADERS",
-      "HEADER", "SYSTEM", "IF", "EQ", "NEQ", "LT", "LTE", "GT", "GTE", "PLUS", "MINUS", "ASTERISK",
+      "HEADER", "SYSTEM", "TIMEOUT", "IF", "EQ", "NEQ", "LT", "LTE", "GT", "GTE", "PLUS", "MINUS", "ASTERISK",


What's our procedure for adding new reserved words? I imagine we do want to note that somewhere.

TIMEOUT is a non-reserved word. This is just a list of all the grammar tokens.

jnh5y · 2022-04-28T15:58:45Z

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/execution/AssertTopicExecutor.java

+      RetryUtil.retryWithBackoff(
+          timeout / RETRY_MS,
+          RETRY_MS,
+          RETRY_MS,


I think this should be timeout. If I'm reading code correctly, there'd only be one retry with the code as-is. (I'm skimming, so happy to be wrong.)

public static void retryWithBackoff( final int maxRetries, final int initialWaitMs, final int maxWaitMs, final Runnable runnable, final Class<?>... passThroughExceptions)

The retryWithBackoff method increases the retry timeout between each retry. It starts at initialWaitMs and then inceases to 2*initialWaitMs and then 4*initialWaitMs etc until it reaches maxWait. So if we keep initialWaitMs == maxWaitMs then it will retry every initialWaitMs ms until we reach maxRetries.

jnh5y · 2022-04-28T15:59:59Z

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/execution/AssertTopicExecutor.java

+
+    if (topicExists) {
+      final int partitions = client.describeTopic(topic).partitions().size();
+      final int replicas = client.describeTopic(topic).partitions().get(0).replicas().size();


Would it be better to factor out the call to client.describeTopic(topic).partitions()? (Not sure if the metadata is cached or what.)

Good call, that's another api call. Done.

jnh5y · 2022-04-28T16:02:19Z

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/execution/AssertTopicExecutor.java

+      final int partitions = client.describeTopic(topic).partitions().size();
+      final int replicas = client.describeTopic(topic).partitions().get(0).replicas().size();
+      final List<String> configErrors = new ArrayList<>();
+      config.forEach((k, v) -> {


As a non-blocking comment, the part of me that's an Scala/FP junkie would offer that it may be an interesting exercise to see if it is easy to switch to a "map" (or maybe a flatmap) and collect the result for the errors.

This code works and is clear, so no obligation to try that.

Also, thank you for collecting all the errors!

Leaving it as is because I tried that end it ended up getting messier :(

jnh5y · 2022-04-28T16:05:24Z

ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/integration/RestIntegrationTestUtil.java

@@ -555,6 +556,7 @@ public static List<String> makeWsRequest(

      mediaType.ifPresent(mt -> headers.add(ACCEPT.toString(), mt));
      contentType.ifPresent(ct -> headers.add(CONTENT_TYPE.toString(), ct));
+      headers.add(EXPIRES.toString(), "Wed, 13 Apr 2022 07:28:00 GMT");


Do we need this header? Also, isn't it expired? I'm confused.

whoops that was from something completely different I was testing

jnh5y

Looks pretty reasonable. Happy to approve after another round of discussions.

jnh5y

Thanks for the discussion! LGTM!

Zara Lim added 2 commits April 26, 2022 15:25

feat: add ASSERT TOPIC command

3f54364

make timeout nonreserved

82cc641

jzaralim requested a review from a team as a code owner April 26, 2022 22:32

Zara Lim added 2 commits April 26, 2022 16:15

checkstyle

6f53c7a

fix unit test

c3bd00b

jzaralim marked this pull request as draft April 27, 2022 08:51

jnh5y reviewed Apr 27, 2022

View reviewed changes

update cli

925dd80

jzaralim marked this pull request as ready for review April 27, 2022 18:57

Zara Lim added 2 commits April 27, 2022 16:25

fix integration test

29cff6d

fix qtt

c1dceac

jnh5y reviewed Apr 28, 2022

View reviewed changes

jnh5y approved these changes Apr 28, 2022

View reviewed changes

address review comments

022417c

jzaralim merged commit cd5254f into master Apr 28, 2022

jzaralim deleted the assert-topic branch April 28, 2022 21:09

jzaralim mentioned this pull request May 12, 2022

docs: add documentation for asserts #9118

Merged

2 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: add ASSERT TOPIC command #9066

feat: add ASSERT TOPIC command #9066

jzaralim commented Apr 26, 2022

jnh5y Apr 27, 2022

jzaralim Apr 27, 2022

jnh5y Apr 28, 2022

jzaralim Apr 28, 2022

jnh5y Apr 27, 2022

jzaralim Apr 27, 2022

jnh5y Apr 27, 2022

jnh5y Apr 28, 2022

jzaralim Apr 28, 2022

jnh5y Apr 28, 2022

jzaralim Apr 28, 2022

jnh5y Apr 28, 2022

jzaralim Apr 28, 2022

jnh5y Apr 28, 2022

jzaralim Apr 28, 2022

jnh5y Apr 28, 2022

jzaralim Apr 28, 2022

jnh5y Apr 28, 2022

jzaralim Apr 28, 2022

jnh5y Apr 28, 2022

jnh5y Apr 28, 2022

jzaralim Apr 28, 2022

jnh5y Apr 28, 2022

jzaralim Apr 28, 2022

jnh5y left a comment

jnh5y left a comment

feat: add ASSERT TOPIC command #9066

feat: add ASSERT TOPIC command #9066

Conversation

jzaralim commented Apr 26, 2022

Description

Testing done

Reviewer checklist

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jnh5y left a comment

Choose a reason for hiding this comment

jnh5y left a comment

Choose a reason for hiding this comment