feat: add rate-limiting to ksql command topic #8809

lct45 · 2022-02-24T16:44:08Z

Description

We recently updated the command topic in #8742 to make sure that it's created with all the proper configs. Now that the command topic can be created on a separate Kafka from the rest of the topics, we want to rate limit the command topic to ensure that it isn't weighing down the kafka it's running on.

Testing done

Tested locally

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 #")

wcarlson5

Couple things. It would be nice to get a test or two nothing crazy. Just so if someone changes related code without knowing about this and break it we will catch it

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

wcarlson5 · 2022-02-25T16:08:45Z

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/CommandRunner.java

@@ -346,6 +354,11 @@ private void executeStatement(final QueuedCommand queuedCommand) {
      if (closed) {
        LOG.info("Execution aborted as system is closing down");
      } else {
+        if (!rateLimiter.tryAcquire()) {
+          throw new KsqlRestException(


How will this error get handled?

And can you a unit test to make sure its gets thrown and handled?

Yeah writing a unit test rn - it will return an error back up to the user

wcarlson5

LGTM

wcarlson5 · 2022-02-25T18:47:53Z

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

@@ -663,6 +663,12 @@
  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_COMMAND_TOPIC_RATE_LIMIT_CONFIG
+      = "ksql.command.topic.rate.limit";
+  public static final double KSQL_COMMAND_TOPIC_RATE_LIMIT_CONFIG_DEFAULT = 10.0;


Nit: Do we need a double or can we use an int?

the library takes a double as permitsPerSecond https://guava.dev/releases/19.0/api/docs/index.html?com/google/common/util/concurrent/RateLimiter.html

Can we set the default much higher (like 1000)? I think 10 is too low for a default and many users may inadvertently hit this.

Also, could you move the config to KsqlRestConfig and follow a similar naming pattern to KSQL_COMMAND_RUNNER_BLOCKED_THRESHHOLD_ERROR_MS and DISTRIBUTED_COMMAND_RESPONSE_TIMEOUT_MS_CONFIG with ksql.server.command prefix? I think we should keep the command topic/command runner configs together and named consistently

@stevenpyzhang I moved it but now it can't be included in the ImmutableProperties list and we get issues when creating the rate limiter. I upped the limit for now, @rodesai WDYT about doing double max value versus 10? seems ok to me

Yes we should definitely disable rate-limiting by default by setting this to MAX_INT. We can enable in cloud as needed.

jnh5y · 2022-02-25T21:23:30Z

Is this going to cause users scripting commands to have a rough time?

lct45 · 2022-02-25T21:38:44Z

@jnh5y correct me if my understanding is wrong, but users who are scripting commands would also be running their own ksql, right? From my understanding Confluent Cloud users can't write a script to create streams / tables unless it issues a curl request, but it seems like that would take enough time to prevent us getting over 10 requests in a second.

If a user is running their own ksql they can configure this through the properties file

jnh5y · 2022-02-25T22:13:17Z

Were I running my own ksql, I think having to set this would be somewhat surprising.

For ccloud, I suppose it depends on how someone was scripting things;).

stevenpyzhang

I thought we wanted to rate limit the number of records written to the command topic? This PR looks like it only rate limits how fast we read the records from the command topic, someone could still write a ton of records to the command topic.

lct45 · 2022-02-28T21:03:29Z

error on the cli:

ksql> create stream c1 as select * from riderlocations; 
Too many writes to the command topic within a 1 second timeframe

wcarlson5 · 2022-02-28T21:04:24Z

@lct45 what happens if you try a batch like this all at once?

Create stream users (A string, B string key) with (kafka_topic='users', VALUE_FORMAT='json', PARTITIONS=1);
create or replace table users_agg0 as select B, count(*) from USERS group by B;
create table users_agg1 as select B, count(*) from USERS group by B;
create table users_agg2 as select B, count(*) from USERS group by B;
create table users_agg3 as select B, count(*) from USERS group by B;
create table users_agg4 as select B, count(*) from USERS group by B;
create table users_agg5 as select B, count(*) from USERS group by B;
create table users_agg6 as select B, count(*) from USERS group by B;
create table users_agg7 as select B, count(*) from USERS group by B;
create table users_agg8 as select B, count(*) from USERS group by B;
create table users_agg9 as select B, count(*) from USERS group by B;
create table users_agg10 as select B, count(*) from USERS group by B;
create table users_agg11 as select B, count(*) from USERS group by B;
create table users_agg12 as select B, count(*) from USERS group by B;
create table users_agg13 as select B, count(*) from USERS group by B;
create table users_agg14 as select B, count(*) from USERS group by B;
create table users_agg15 as select B, count(*) from USERS group by B;
create table users_agg16 as select B, count(*) from USERS group by B;
create table users_agg17 as select B, count(*) from USERS group by B;
create table users_agg18 as select B, count(*) from USERS group by B;

lct45 · 2022-02-28T21:10:25Z

@wcarlson5 it executed all the statements with no errors (I did it as one block from the CLI)

stevenpyzhang · 2022-02-28T21:59:22Z

...st-app/src/test/java/io/confluent/ksql/rest/server/computation/DistributingExecutorTest.java

+  public void shouldThrowIfRateLimitHit() {
+    // Given:
+    final DistributingExecutor rateLimitedDistributor = new DistributingExecutor(
+      new KsqlConfig(ImmutableMap.of("ksql.command.topic.rate.limit", 1.0)),


Nit: use the const in KsqlConfig

rodesai · 2022-03-01T17:18:54Z

...b-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/DistributingExecutor.java

+    if (!rateLimiter.tryAcquire()) {
+      throw new KsqlRestException(
+        Errors.tooManyRequests(
+          "Too many writes to the command topic within a 1 second timeframe"


nit: change this to "DDL/DML rate is crossing the configured rate limit of statements/second"

rodesai · 2022-03-01T17:23:39Z

...b-rest-app/src/main/java/io/confluent/ksql/rest/server/computation/DistributingExecutor.java

@@ -196,6 +203,13 @@ public StatementExecutorResponse execute(
          statement.getStatementText()), e);
    }

+    if (!rateLimiter.tryAcquire()) {


I think it would make sense to tryAcquire with a delay of a second here. This way we don't fail scripts that serially try to enqueue a large number of commands.

@rodesai just added -> will you confirm this is what you were thinking?

rodesai

LGTM

lct45 · 2022-03-02T15:26:57Z

https://jenkins.confluent.io/job/confluentinc-pr/job/ksql/job/PR-8809/16/ had a green build

feat: add rate-limiting to ksql command topic

b26d7b9

lct45 force-pushed the rate-limiting branch from caca6f1 to daebdfe Compare February 24, 2022 22:00

make rate limit quota set by config

fba9fe4

lct45 force-pushed the rate-limiting branch from daebdfe to fba9fe4 Compare February 25, 2022 15:21

make rate limiting return an error

67b1011

lct45 marked this pull request as ready for review February 25, 2022 15:42

lct45 requested a review from a team as a code owner February 25, 2022 15:42

wcarlson5 reviewed Feb 25, 2022

View reviewed changes

lct45 requested review from rodesai and JSeb225 February 25, 2022 18:27

wcarlson5 approved these changes Feb 25, 2022

View reviewed changes

test + review

5b543be

lct45 force-pushed the rate-limiting branch from ac6461c to 5b543be Compare February 25, 2022 21:38

stevenpyzhang reviewed Feb 25, 2022

View reviewed changes

lct45 force-pushed the rate-limiting branch from 8eb3fe9 to d63e625 Compare February 28, 2022 15:35

move rate limiting to commandstore

6d168be

lct45 force-pushed the rate-limiting branch from d63e625 to 6d168be Compare February 28, 2022 18:01

stevenpyzhang approved these changes Feb 28, 2022

View reviewed changes

move rate limiting to distributing executor

e2465ba

lct45 force-pushed the rate-limiting branch from 654aa6d to e2465ba Compare February 28, 2022 21:02

stevenpyzhang reviewed Feb 28, 2022

View reviewed changes

stevenpyzhang approved these changes Feb 28, 2022

View reviewed changes

stevenpyzhang self-requested a review February 28, 2022 22:04

move config to rest config

3aedfb1

rodesai reviewed Mar 1, 2022

View reviewed changes

lct45 requested a review from rodesai March 1, 2022 17:37

rodesai approved these changes Mar 1, 2022

View reviewed changes

rohan's review

181d762

lct45 force-pushed the rate-limiting branch from eb13d22 to 181d762 Compare March 1, 2022 19:58

lct45 merged commit b2f1540 into confluentinc:master Mar 2, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: add rate-limiting to ksql command topic #8809

feat: add rate-limiting to ksql command topic #8809

lct45 commented Feb 24, 2022 •

edited

Loading

wcarlson5 left a comment

wcarlson5 Feb 25, 2022

lct45 Feb 25, 2022

wcarlson5 left a comment

wcarlson5 Feb 25, 2022

lct45 Feb 25, 2022

stevenpyzhang Feb 28, 2022

stevenpyzhang Feb 28, 2022

lct45 Mar 1, 2022 •

edited

Loading

rodesai Mar 1, 2022

jnh5y commented Feb 25, 2022

lct45 commented Feb 25, 2022

jnh5y commented Feb 25, 2022

stevenpyzhang left a comment

lct45 commented Feb 28, 2022

wcarlson5 commented Feb 28, 2022 •

edited

Loading

lct45 commented Feb 28, 2022

stevenpyzhang Feb 28, 2022

rodesai Mar 1, 2022

lct45 Mar 1, 2022

rodesai Mar 1, 2022

lct45 Mar 1, 2022

rodesai left a comment

lct45 commented Mar 2, 2022

feat: add rate-limiting to ksql command topic #8809

feat: add rate-limiting to ksql command topic #8809

Conversation

lct45 commented Feb 24, 2022 • edited Loading

Description

Testing done

Reviewer checklist

wcarlson5 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wcarlson5 left a 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

lct45 Mar 1, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jnh5y commented Feb 25, 2022

lct45 commented Feb 25, 2022

jnh5y commented Feb 25, 2022

stevenpyzhang left a comment

Choose a reason for hiding this comment

lct45 commented Feb 28, 2022

wcarlson5 commented Feb 28, 2022 • edited Loading

lct45 commented Feb 28, 2022

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

rodesai left a comment

Choose a reason for hiding this comment

lct45 commented Mar 2, 2022

lct45 commented Feb 24, 2022 •

edited

Loading

lct45 Mar 1, 2022 •

edited

Loading

wcarlson5 commented Feb 28, 2022 •

edited

Loading