fix: update error msg (#8221)

* fix: update error msg * fix: update name * fix: tests
confluentinc · Oct 6, 2021 · 8c30780 · 8c30780
1 parent 68f05a0
commit 8c30780
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 6 deletions.
diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/topic/TopicCreateInjector.java b/ksqldb-engine/src/main/java/io/confluent/ksql/topic/TopicCreateInjector.java
@@ -110,7 +110,7 @@ private ConfiguredStatement<? extends CreateSource> injectForCreateSource(
     } else if (!properties.getPartitions().isPresent()) {
       final CreateSource example = createSource.copyWith(
           createSource.getElements(),
-          properties.withPartitionsAndReplicas(2, (short) 1));
+          properties.withPartitions(2));
       throw new KsqlException(
           "Topic '" + topicName + "' does not exist. If you want to create a new topic for the "
               + "stream/table please re-run the statement providing the required '"

diff --git a/ksqldb-engine/src/test/java/io/confluent/ksql/topic/TopicCreateInjectorTest.java b/ksqldb-engine/src/test/java/io/confluent/ksql/topic/TopicCreateInjectorTest.java
@@ -461,7 +461,7 @@ public void shouldHaveSuperUsefulErrorMessageIfCreateWithNoPartitions() {
         + "stream/table please re-run the statement providing the required 'PARTITIONS' "
         + "configuration in the WITH clause (and optionally 'REPLICAS'). For example: "
         + "CREATE STREAM FOO (FOO STRING) "
-        + "WITH (KAFKA_TOPIC='doesntexist', PARTITIONS=2, REPLICAS=1, VALUE_FORMAT='avro');"));
+        + "WITH (KAFKA_TOPIC='doesntexist', PARTITIONS=2, VALUE_FORMAT='avro');"));
   }
 
   private ConfiguredStatement<?> givenStatement(final String sql) {

diff --git a/...parser/src/main/java/io/confluent/ksql/parser/properties/with/CreateSourceProperties.java b/...parser/src/main/java/io/confluent/ksql/parser/properties/with/CreateSourceProperties.java
@@ -204,13 +204,11 @@ public CreateSourceProperties withSchemaIds(
     return new CreateSourceProperties(originals, durationParser);
   }
 
-  public CreateSourceProperties withPartitionsAndReplicas(
-      final int partitions,
-      final short replicas
+  public CreateSourceProperties withPartitions(
+      final int partitions
   ) {
     final Map<String, Literal> originals = props.copyOfOriginalLiterals();
     originals.put(CommonCreateConfigs.SOURCE_NUMBER_OF_PARTITIONS, new IntegerLiteral(partitions));
-    originals.put(CommonCreateConfigs.SOURCE_NUMBER_OF_REPLICAS, new IntegerLiteral(replicas));
 
     return new CreateSourceProperties(originals, durationParser);
   }