chore: inject schema using schema id for cas #8441

lihaosky · 2021-12-02T19:31:42Z

Description

Inject schema for cas statement if schema id is provided.
Update QTT test to use SchemaRegisterInjector as well.
Disabled a few failing QTT tests as it also fails in current master branch (Decimal not working with schema registry #8465)

Testing done

Unit test
QTT test

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

spena · 2021-12-03T22:53:38Z

ksqldb-engine/src/main/java/io/confluent/ksql/schema/ksql/inference/DefaultSchemaInjector.java

 * <ul>
- * <li>The statement is a CT/CS.</li>
- * <li>The statement does not defined any key columns.</li>
+ * <li>The statement does not defined any key columns or has KEY_SCHEMA_ID property set</li>


If KEY_SCHEMA_ID is set, then no key columns must be defined either, right? Could you update the line?

spena · 2021-12-03T22:53:59Z

ksqldb-engine/src/main/java/io/confluent/ksql/schema/ksql/inference/DefaultSchemaInjector.java

 * <ul>
- * <li>The statement is a CT/CS.</li>
- * <li>The statement does not defined any key columns.</li>
+ * <li>The statement does not defined any key columns or has KEY_SCHEMA_ID property set</li>
 * <li>The key format of the statement supports schema inference.</li>
 * </ul>
 * And similarly for value columns.


VALUE_SCHEMA_ID is the property in the case of value columns, right? Can you update the doc?

spena · 2021-12-03T23:11:50Z

...src/main/java/io/confluent/ksql/schema/ksql/inference/SchemaRegistryTopicSchemaSupplier.java

        id = srClient.getLatestSchemaMetadata(subject).getId();
      }

      final ParsedSchema schema = srClient.getSchemaBySubjectAndId(subject, id);
-      return fromParsedSchema(topicName, id, schema, expectedFormat, serdeFeatures, isKey,
-          schemaId.isPresent());
+      return fromParsedSchema(topicName, id, schemaId, schema, expectedFormat, serdeFeatures,


id and schemaId are the same. Can this become into one parameter?

Will take a look how to refactor

Actually schemaId is optional and is used to indicate whether schemaId is provided by user to fetched from latest schema. Maybe I can refactor to call it isIdProvided

spena · 2021-12-06T16:58:06Z

ksqldb-serde/src/test/java/io/confluent/ksql/serde/protobuf/KsqlProtobufSerializerTest.java

+        containsString("Error serializing message to topic: bob. "
+            + "Failed to serialize Protobuf data from topic bob"));


is there a way to know what's the real error? Is this a partial error btw?

This is actually dependent PR's code and removed. Let me rebase this PR...

spena · 2021-12-07T20:22:30Z

ksqldb-engine/src/main/java/io/confluent/ksql/schema/ksql/inference/SchemaRegisterInjector.java

+            PersistenceSchema.from(schema, serdeFeatures)
+        );
+        int id = SchemaRegistryUtil.registerSchema(srClient, parsedSchema, topic, subject, isKey);
+        System.out.print("h");


is this print an accident?

Will remove

spena · 2021-12-07T20:22:51Z

ksqldb-engine/src/main/java/io/confluent/ksql/schema/ksql/inference/SchemaRegisterInjector.java

+        final ParsedSchema parsedSchema = translator.toParsedSchema(
+            PersistenceSchema.from(schema, serdeFeatures)
+        );
+        int id = SchemaRegistryUtil.registerSchema(srClient, parsedSchema, topic, subject, isKey);


what do you do with the id?

Will remove

spena · 2021-12-07T20:24:15Z

ksqldb-functional-tests/src/main/java/io/confluent/ksql/test/tools/TestExecutorUtil.java

+          int id = srClient.register(KsqlConstants.getSRSubject(topic.getName(), true), schema);
+          System.out.print("h");


where is the id use? What about the print call?

Will remove

spena · 2021-12-07T20:24:48Z

ksqldb-functional-tests/src/test/java/io/confluent/ksql/test/QueryTranslationTest.java

    this.testCase = requireNonNull(testCase, "testCase");
  }

  @Test
  public void shouldBuildAndExecuteQueries() {
+    if (!testCase.getName().contains("DECIMAL - key - inference - default precision")) {


Is this usd for debugging?

Yeah. Will remove

spena · 2021-12-07T20:31:43Z

ksqldb-functional-tests/src/test/resources/query-validation-tests/kafka.json

@@ -7,7 +7,7 @@
        "CREATE STREAM INPUT (K STRING KEY, foo INT) WITH (WRAP_SINGLE_VALUE=true, kafka_topic='input', value_format='KAFKA');"
      ],
      "expectedException": {
-        "type": "io.confluent.ksql.util.KsqlStatementException",
+        "type": "io.confluent.ksql.util.KsqlException",


Can the KsqlStatementException stay? There is a check at the end of the KsqlResource.handleKsqlStatements() method that returns a different value depending on the exception.

This question applies for the rest of the tests that changes to this new exception.

Sure. Will catch KsqlException and rethrow

spena

looks good

lihaosky requested review from spena and mjsax December 2, 2021 19:31

lihaosky requested a review from a team as a code owner December 2, 2021 19:31

lihaosky force-pushed the schema-id-cas branch from 5e6f5a5 to ce28161 Compare December 2, 2021 21:30

spena reviewed Dec 3, 2021

View reviewed changes

spena reviewed Dec 6, 2021

View reviewed changes

lihaosky force-pushed the schema-id-cas branch 4 times, most recently from 5412f7e to 14c925e Compare December 7, 2021 20:20

spena reviewed Dec 7, 2021

View reviewed changes

spena approved these changes Dec 7, 2021

View reviewed changes

lihaosky force-pushed the schema-id-cas branch from a12401a to 7b69541 Compare December 8, 2021 01:27

lihaosky added 7 commits December 7, 2021 22:50

chore: inject raw schema for c*as command

f35a307

fix: address comments and rebase

c59c25c

fix: checkstyle

e88c65d

test: qtt test

68d7b9b

test: fix test

0d7bb0b

test: regen qtt

f29ece6

test: update qtt

85415a7

lihaosky force-pushed the schema-id-cas branch from 7b69541 to 85415a7 Compare December 8, 2021 07:22

lihaosky merged commit a60bb21 into confluentinc:master Dec 8, 2021

lihaosky deleted the schema-id-cas branch December 8, 2021 16:22

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

chore: inject schema using schema id for cas #8441

chore: inject schema using schema id for cas #8441

lihaosky commented Dec 2, 2021 •

edited

Loading

spena Dec 3, 2021

spena Dec 3, 2021

spena Dec 3, 2021

lihaosky Dec 3, 2021

lihaosky Dec 6, 2021

spena Dec 6, 2021

lihaosky Dec 6, 2021

spena Dec 7, 2021

lihaosky Dec 7, 2021

spena Dec 7, 2021

lihaosky Dec 7, 2021

spena Dec 7, 2021

lihaosky Dec 7, 2021

spena Dec 7, 2021

lihaosky Dec 7, 2021

spena Dec 7, 2021

lihaosky Dec 7, 2021

spena left a comment

		containsString("Error serializing message to topic: bob. "
		+ "Failed to serialize Protobuf data from topic bob"));

		int id = srClient.register(KsqlConstants.getSRSubject(topic.getName(), true), schema);
		System.out.print("h");

chore: inject schema using schema id for cas #8441

chore: inject schema using schema id for cas #8441

Conversation

lihaosky commented Dec 2, 2021 • edited Loading

Description

Testing done

Reviewer checklist

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

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

spena left a comment

Choose a reason for hiding this comment

lihaosky commented Dec 2, 2021 •

edited

Loading