KSQL-13911 | Auto-migrate exactly_once to exactly_once_v2

Author: vi-tayal

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

@@ -30,6 +30,7 @@
 import io.confluent.ksql.rest.server.resources.IncompatibleKsqlCommandVersionException;
 import io.confluent.ksql.statement.ConfiguredStatement;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -112,6 +113,10 @@ public String getStatement() {
     return statement;
   }
 
+  /**
+   * Returns the overwrite properties for JSON serialization to command topic.
+   * This preserves the original values as they were stored.
+   */
   @JsonProperty("streamsProperties")
   @SuppressFBWarnings(
       value = "EI_EXPOSE_REP",
@@ -121,6 +126,28 @@ public Map<String, Object> getOverwriteProperties() {
     return PropertiesUtil.coerceTypes(overwriteProperties, true);
   }
 
+  /**
+   * Returns the overwrite properties for runtime use, with legacy values migrated.
+   * This is what should be used when executing queries.
+   */
+  public Map<String, Object> getOverwritePropertiesForExecution() {
+    // Create mutable copy and migrate BEFORE coerceTypes validation
+    final Map<String, Object> mutableProps = new HashMap<>(overwriteProperties);
+    migrateLegacyProcessingGuarantee(mutableProps);
+    // Now validate and coerce the migrated values
+    return PropertiesUtil.coerceTypes(mutableProps, true);
+  }
+
+  private static void migrateLegacyProcessingGuarantee(final Map<String, Object> properties) {
+    final Object guarantee = properties.get("processing.guarantee");
+    if (guarantee != null) {
+      final String guaranteeStr = guarantee.toString();
+      if ("exactly_once".equals(guaranteeStr)) {
+        properties.put("processing.guarantee", "exactly_once_v2");
+      }
+    }
+  }
+
   @SuppressFBWarnings(
       value = "EI_EXPOSE_REP",
       justification = "originalProperties is unmodifiableMap()"

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

@@ -232,7 +232,7 @@ private void executePlan(
     final KsqlConfig mergedConfig = buildMergedConfig(command);
     final ConfiguredKsqlPlan configured = ConfiguredKsqlPlan.of(
         plan,
-        SessionConfig.of(mergedConfig, command.getOverwriteProperties())
+        SessionConfig.of(mergedConfig, command.getOverwritePropertiesForExecution())
     );
     putStatus(
         commandId,
@@ -321,7 +321,7 @@ private void executeStatement(
 
   private KsqlConfig buildMergedConfig(final Command command) {
     return ksqlEngine.getKsqlConfig()
-        .overrideBreakingConfigsWithOriginalValues(command.getOriginalProperties());
+        .overrideBreakingConfigsWithOriginalValues(command.getOverwritePropertiesForExecution());
   }
 
   private void pauseQuery(final PreparedStatement<PauseQuery> pauseQuery) {

ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/computation/CommandTest.java

@@ -135,4 +135,47 @@ public void shouldCoerceProperties() {
         Matchers.equalTo((short) 3)
     );
   }
+
+  @Test
+  public void shouldNotMigrateModernProcessingGuaranteeValues() {
+    // Given: Command with modern processing guarantee values
+    final Command command1 = new Command(
+        "CREATE STREAM test AS SELECT * FROM source;",
+        ImmutableMap.of("processing.guarantee", "exactly_once_v2"),
+        Collections.emptyMap(),
+        Optional.empty()
+    );
+    final Command command2 = new Command(
+        "CREATE STREAM test AS SELECT * FROM source;",
+        ImmutableMap.of("processing.guarantee", "at_least_once"),
+        Collections.emptyMap(),
+        Optional.empty()
+    );
+
+    // When: Getting overwrite properties
+    final Map<String, Object> properties1 = command1.getOverwriteProperties();
+    final Map<String, Object> properties2 = command2.getOverwriteProperties();
+
+    // Then: Should remain unchanged
+    assertThat(properties1.get("processing.guarantee"), equalTo("exactly_once_v2"));
+    assertThat(properties2.get("processing.guarantee"), equalTo("at_least_once"));
+  }
+
+  @Test
+  public void shouldHandleCommandWithNoProcessingGuarantee() {
+    // Given: Command without processing.guarantee property
+    final Command command = new Command(
+        "CREATE STREAM test AS SELECT * FROM source;",
+        ImmutableMap.of("num.stream.threads", 4),
+        Collections.emptyMap(),
+        Optional.empty()
+    );
+
+    // When: Getting overwrite properties
+    final Map<String, Object> properties = command.getOverwriteProperties();
+
+    // Then: Should not add processing.guarantee
+    assertThat(properties.containsKey("processing.guarantee"), is(false));
+    assertThat(properties.get("num.stream.threads"), equalTo(4));
+  }
 }