fix: disallow requests to /inserts-stream if insert values disabled (#…

…5592)
confluentinc · Jun 10, 2020 · e277f25 · e277f25
1 parent 174756d
commit e277f25
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 1 deletion.
diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/engine/InsertValuesExecutor.java b/ksqldb-engine/src/main/java/io/confluent/ksql/engine/InsertValuesExecutor.java
@@ -249,7 +249,7 @@ private void throwIfDisabled(final KsqlConfig config) {
 
     if (canBeDisabledByConfig && !isEnabled) {
       throw new KsqlException("The server has disabled INSERT INTO ... VALUES functionality. "
-          + "To enable it, restart your KSQL-server "
+          + "To enable it, restart your ksqlDB server "
           + "with 'ksql.insert.into.values.enabled'=true");
     }
   }

diff --git a/ksqldb-rest-app/src/main/java/io/confluent/ksql/api/impl/InsertsStreamEndpoint.java b/ksqldb-rest-app/src/main/java/io/confluent/ksql/api/impl/InsertsStreamEndpoint.java
@@ -15,6 +15,7 @@
 
 package io.confluent.ksql.api.impl;
 
+import static io.confluent.ksql.rest.Errors.ERROR_CODE_BAD_REQUEST;
 import static io.confluent.ksql.rest.Errors.ERROR_CODE_BAD_STATEMENT;
 
 import io.confluent.ksql.api.server.InsertResult;
@@ -53,6 +54,12 @@ public InsertsStreamSubscriber createInsertsSubscriber(final String target,
       final WorkerExecutor workerExecutor,
       final ServiceContext serviceContext) {
     VertxUtils.checkIsWorker();
+    if (!ksqlConfig.getBoolean(KsqlConfig.KSQL_INSERT_INTO_VALUES_ENABLED)) {
+      throw new KsqlApiException("The server has disabled INSERT INTO ... VALUES functionality. "
+          + "To enable it, restart your ksqlDB server "
+          + "with 'ksql.insert.into.values.enabled'=true",
+          ERROR_CODE_BAD_REQUEST);
+    }
     final DataSource dataSource = getDataSource(ksqlEngine.getMetaStore(),
         SourceName.of(target));
     if (dataSource.getDataSourceType() == DataSourceType.KTABLE) {