fix: default timestamp extractor override is not working #3176

spena · 2019-08-06T14:04:27Z

Description

The use of ksql.streams.default.timestamp.extractor when creating a stream/table is not working. The new value is persisted in the command topic, but KSQL always use a default FailOnInvalidTimestamp

This patch fixes KSQL so it honors the new default specified.

Testing done

Added unit tests.
Verified a stream created with 'ksql.streams.default.timestamp.extractor'='org.apache.kafka.streams.processor.WallclockTimestampExtractor' results in queries displaying the current Wallclock time.
Verified a stream created with no timestamp extractor uses the default FailedOnInvalidTimestamp, which displays the timestamp it was used during the insertion of the row.

To test it, I had to call the REST api instead of the CLI because I couldn't make the CLI to recognize a different timestamp extractor:

curl -X POST http://localhost:8088/ksql -H 'content-type: application/vnd.ksql.v1+json; charset=utf-8' -d "{\"ksql\":\"create stream t1 (id int, name string) with (kafka_topic='t1', value_format='json');\", \"streamsProperties\": { \"ksql.streams.default.timestamp.extractor\": \"org.apache.kafka.streams.processor.WallclockTimestampExtractor\" }}"

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

agavra

Overall this LGTM, but I'm trying to understand why doesn't it work in the CLI?

agavra · 2019-08-06T15:46:26Z

ksql-engine/src/main/java/io/confluent/ksql/ddl/commands/CreateSourceCommand.java

+
+      return (TimestampExtractor) defaultTimestampExtractor.newInstance();
+    } catch (final Exception e) {
+      // TODO: Need to log invalid timestamp configuration here?


why not violently fail (e.g. raise exception) here? I think I'd rather have my command fail than do something which wasn't what i defined

I forgot to remove that line. I left it as a reminder for me to do something there before submitting the PR.

Anyway, you're right, I should fail instead of just logging the error. I submitted another PR (this time for 5.3.x branch instead of master) which fails in that case.

agavra · 2019-08-06T15:47:12Z

ksql-engine/src/test/java/io/confluent/ksql/ddl/commands/CreateSourceCommandTest.java

@@ -230,6 +233,51 @@ public void shouldBuildSerdeOptions() {
    assertThat(cmd.getSerdeOptions(), is(SOME_SERDE_OPTIONS));
  }

+  @Test


can we add a QTT for this? we should be able to pass it in the property overrides for a command

Just added one for this.

agavra

Thanks @spena! LGTM

blueedgenick · 2019-08-06T23:48:20Z

seems very strange to me also that this doesn't work using SET from the CLI. Arguably the fix is incomplete if this doesn't work - is there at least another issue opened up to track that ?

hjafarpour · 2019-08-07T18:12:26Z

I agree with @blueedgenick, this should work from CLI too. The problem with SET in CLI is that we validate the config properties in the CLI side before sending the requests to the server and the validation for default.timestamp.extractor property uses Class.forName and if you don't have the class in the class path in client side this will through exception.
You can either make sure that the streams classes are available in the CLI side or not validate this config property in the CLI side.

spena · 2019-08-07T18:28:07Z

@agavra I submitted another commit to fix the issue when the CLI fails sending a REST request that contains a Class value in the stream properties field.

spena · 2019-08-07T21:12:25Z

@hjafarpour I decided to convert the Class values of the stream properties to String values. For some reason, Jackson does not detect the class in the classpath when creating the JSON object. The jar is in the classpath, in fact, the Class is available in the streams properties as Class. But when doing the JSON conversion, Jackson fails.

The code is working now. If you're fine with it, then I can merge it later.

hjafarpour

Thanks @spena
Added a few comments.

hjafarpour · 2019-08-07T21:00:19Z

...-common/src/main/java/io/confluent/ksql/util/timestamp/TimestampExtractionPolicyFactory.java

@@ -31,9 +37,23 @@ public static TimestampExtractionPolicy create(
      final KsqlSchema schema,
      final Optional<String> timestampColumnName,
      final Optional<String> timestampFormat
+  ) {


This was used only in one place, LogicalPlanner. Why not remove this method and just use the next one?

I'll do that, thanks.

hjafarpour · 2019-08-07T21:31:00Z

...mon/src/test/java/io/confluent/ksql/util/timestamp/TimestampExtractionPolicyFactoryTest.java

+    assertThat(result.create(0), instanceOf(FailOnInvalidTimestamp.class));
+  }
+
+  @Test(expected = ConfigException.class)


Why not use ExpectedException so you can also check the correctness of the error messages?

I followed how other tests cases of this class verify error exceptions.
Btw, What is our preference? Should I modify the other tests cases with the ExpectdException?

ExpectdException would be preferred. Check KsqlRequestTest for examples.

hjafarpour · 2019-08-07T21:31:12Z

...mon/src/test/java/io/confluent/ksql/util/timestamp/TimestampExtractionPolicyFactoryTest.java

+        );
+  }
+
+  @Test(expected = KsqlException.class)


Ditto here, use ExpectedException

hjafarpour · 2019-08-07T21:33:31Z

ksql-rest-app/src/main/java/io/confluent/ksql/rest/entity/KsqlRequest.java

@@ -46,7 +47,7 @@ public KsqlRequest(
    this.ksql = ksql == null ? "" : ksql;
    this.streamsProperties = streamsProperties == null
        ? Collections.emptyMap()
-        : Collections.unmodifiableMap(new HashMap<>(streamsProperties));
+        : Collections.unmodifiableMap(new HashMap<>(serializeClassValues(streamsProperties)));


It would be good to add a test for this.

hjafarpour

LGTM.

spena added the bug label Aug 6, 2019

spena added this to the 5.4 milestone Aug 6, 2019

spena requested review from hjafarpour and a team August 6, 2019 14:04

spena self-assigned this Aug 6, 2019

agavra reviewed Aug 6, 2019

View reviewed changes

agavra requested a review from a team August 6, 2019 15:47

spena force-pushed the override_extraction_policy branch from 7970a25 to 4f2d7d5 Compare August 6, 2019 18:43

spena changed the base branch from master to 5.3.x August 6, 2019 18:43

spena force-pushed the override_extraction_policy branch from 4f2d7d5 to ae653d6 Compare August 6, 2019 18:47

agavra approved these changes Aug 6, 2019

View reviewed changes

agavra requested a review from a team August 6, 2019 20:05

hjafarpour reviewed Aug 7, 2019

View reviewed changes

spena added 7 commits August 8, 2019 09:18

refactor: pass KsqlConfig to the CreateSourceCommand class

c046741

fix: default timestamp extractor override is not working

c5a6592

fix: tests failing with DLS_DEAD_LOCAL_STORE

1d7d68f

fix: cli fails when sending properties with Class values

e28caf9

test: add test to serialize Class values on KsqlRequest

eea40bf

refactor: call new constructor on LogicalPlanner

5ec3402

test: use expectedException on test cases

598c27a

spena force-pushed the override_extraction_policy branch from 5b72f5f to 598c27a Compare August 8, 2019 14:25

hjafarpour approved these changes Aug 8, 2019

View reviewed changes

spena merged commit d1db07b into confluentinc:5.3.x Aug 8, 2019

spena deleted the override_extraction_policy branch August 8, 2019 19:14

apurvam mentioned this pull request Aug 13, 2019

KSQL streams default.timestamp.extractor override is not working #3164

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix: default timestamp extractor override is not working #3176

fix: default timestamp extractor override is not working #3176

spena commented Aug 6, 2019

agavra left a comment

agavra Aug 6, 2019

spena Aug 6, 2019

agavra Aug 6, 2019

spena Aug 6, 2019

agavra left a comment

blueedgenick commented Aug 6, 2019

hjafarpour commented Aug 7, 2019

spena commented Aug 7, 2019

spena commented Aug 7, 2019

hjafarpour left a comment

hjafarpour Aug 7, 2019

spena Aug 7, 2019

spena Aug 7, 2019

hjafarpour Aug 7, 2019

spena Aug 7, 2019

hjafarpour Aug 7, 2019

spena Aug 8, 2019

hjafarpour Aug 7, 2019

spena Aug 8, 2019

hjafarpour Aug 7, 2019

spena Aug 7, 2019

hjafarpour left a comment

fix: default timestamp extractor override is not working #3176

fix: default timestamp extractor override is not working #3176

Conversation

spena commented Aug 6, 2019

Description

Testing done

Reviewer checklist

agavra left a 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

agavra left a comment

Choose a reason for hiding this comment

blueedgenick commented Aug 6, 2019

hjafarpour commented Aug 7, 2019

spena commented Aug 7, 2019

spena commented Aug 7, 2019

hjafarpour left a 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

hjafarpour left a comment

Choose a reason for hiding this comment