fix:confluentinc#3347 If the table has an key and a value for it has …

…been provided in an insert then make sure rowkey gets the string form of it
purplefox · Sep 18, 2019 · d49f282 · d49f282
1 parent 62f165b
commit d49f282
Showing 1 changed file with 6 additions and 2 deletions.
diff --git a/ksql-engine/src/main/java/io/confluent/ksql/engine/InsertValuesExecutor.java b/ksql-engine/src/main/java/io/confluent/ksql/engine/InsertValuesExecutor.java
@@ -323,8 +323,12 @@ private static void handleExplicitKeyField(
       final Object rowKeyValue = values.get(SchemaUtil.ROWKEY_NAME);
 
       if (keyValue != null ^ rowKeyValue != null) {
-        values.putIfAbsent(key, rowKeyValue);
-        values.putIfAbsent(SchemaUtil.ROWKEY_NAME, keyValue);
+        if (keyValue == null) {
+          values.putIfAbsent(key, rowKeyValue);
+        } else {
+          // Note, ROWKEY must always be a String
+          values.putIfAbsent(SchemaUtil.ROWKEY_NAME, keyValue.toString());
+        }
       } else if (!Objects.equals(keyValue, rowKeyValue)) {
         throw new KsqlException(
             String.format(