Use source partition count and replica count as default for sink topics. #1930

hjafarpour · 2018-09-21T22:53:06Z

Description

Currently we use the defaults from the KsqlConfig for number of partitions (default is 4) and number of replicas (default is 1) for sink topics if they are not specified in WITH clause.
This PR changes this behavior by using the source partition count and replica count for sink topic if we don't have them specified in the WITH clause.
In case of JOIN we will use the left hand side stream/table topic properties.

Testing done

Unit tests were added.

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

big-andy-coates

Thanks @hjafarpour - left one comment.

ksql-engine/src/main/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNode.java

ksql-engine/src/test/java/io/confluent/ksql/KsqlEngineTest.java

ksql-engine/src/test/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNodeTest.java

rodesai · 2018-09-25T06:23:02Z

ksql-engine/src/main/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNode.java

    final int partitions = (Integer) outputProperties.getOrDefault(
        KsqlConfig.SINK_NUMBER_OF_PARTITIONS_PROPERTY,
-        ksqlConfig.getInt(KsqlConfig.SINK_NUMBER_OF_PARTITIONS_PROPERTY));
+        sourceTopicProperties.getPartitions());


I think our admin client wrapper will validate that this number matches the partition count of the underlying topic. We should disable that check when creating sink topics.

I updated it as we talked offline.

big-andy-coates · 2018-09-26T21:17:22Z

@hjafarpour, in case you've missed it... #1930 (comment)

big-andy-coates · 2018-10-18T13:49:46Z

@hjafarpour any thoughts on this?

…Props-If-Not-Specified

hjafarpour · 2018-12-20T00:40:53Z

@big-andy-coates I updated this PR. Also refactored the test to use Mockito.

big-andy-coates

Thanks @hjafarpour, let's some comments.

ksql-engine/src/main/java/io/confluent/ksql/util/SourceTopicProperties.java

ksql-engine/src/main/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNode.java

ksql-engine/src/test/java/io/confluent/ksql/KsqlEngineTest.java

ksql-engine/src/test/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNodeTest.java

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

…Props-If-Not-Specified

vcrfxia

Thanks @hjafarpour ! LGTM, minus a few minor things in comments.

ksql-engine/src/main/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNode.java

ksql-engine/src/test/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNodeTest.java

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

vcrfxia · 2019-01-08T21:53:05Z

ksql-engine/src/test/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNodeTest.java


 public class KsqlStructuredDataOutputNodeTest {
-  private final KafkaTopicClient topicClient = EasyMock.createNiceMock(KafkaTopicClient.class);
+
+  private KafkaTopicClient topicClient;


nit: move below static constants, e.g., to line 117.

ksql-engine/src/test/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNodeTest.java

…Props-If-Not-Specified

hjafarpour · 2019-01-16T22:33:02Z

@vcrfxia please check this: #1930 (comment)

…Props-If-Not-Specified

big-andy-coates

Thanks @hjafarpour,

It seems that KsqlConfig.SINK_NUMBER_OF_PARTITIONS_PROPERTY and SINK_NUMBER_OF_REPLICAS_PROPERTY are still used in the code. But I thought this PR was change the behaviour to copy the partition count and rf from the source. So in what situations are these two config items still used?

Also, it's unclear to me how we're maintaining backwards compatibility with existing running queries i.e. say the source topic has 8 partitions, and the sink was created with the default 4, then once the user upgrades won't KSQL complain because it expects the sink to have 8 partitions?

Otherwise the code is looking good.

Thanks,

Andy

…Props-If-Not-Specified

hjafarpour

@big-andy-coates I added backward compatibility using a new config property, sink.legacy.properties. Also since we have backward compatibility now we still use KsqlConfig.SINK_NUMBER_OF_PARTITIONS_PROPERTY and SINK_NUMBER_OF_REPLICAS_PROPERTY.

ksql-engine/src/main/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNode.java

apurvam · 2019-02-06T06:43:19Z

ksql-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java

@@ -93,6 +93,14 @@
      + "'CREATE STREAM S AS ...' will create a topic 'thing-S', where as the statement "
      + "'CREATE STREAM S WITH(KAFKA_TOPIC = 'foo') AS ...' will create a topic 'foo'.";

+  public static final String KSQL_SINK_TOPIC_PROPERTIES_LEGACY_CONFIG =
+      KSQL_CONFIG_PROPERTY_PREFIX + "sink.legacy.properties";


I would rather call this something like sink.topic.static.default.properties. It expresses the semantics more cleanly. legacy doesn't mean much over time.

I think we should consider removing KSQL_SINK_TOPIC_PROPERTIES_LEGACY_CONFIG and instead moving both SINK_NUMBER_OF_PARTITIONS_PROPERTY and SINK_NUMBER_OF_REPLICAS_PROPERTY to be CompatibilityBreakingConfigDef instances. The new default for these properties could be null, meaning they are not set. The documentation should say they should not be set going forward. (Ideally, we need a way to hide these internal configs from users)

Going forward, we don't really want people setting either of these properties in their config or in their SET statements, i.e. we're not looking to support "static counts" as a feature. We want them to age out over time. Hence moving them out of the main config and into CompatibilityBreakingConfigDef. This also avoids adding a new config to manage something we'd like to get rid of.

apurvam · 2019-02-06T06:43:27Z

ksql-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java

+  private static final String KSQL_SINK_TOPIC_PROPERTIES_LEGACY_DOCS = "Use the default sink topic"
+      + " properties instead of inferring from source when the properties are not set in the "
+      + " WITH clause. Until version 5.1.x if the partition number and replicas were not specified"
+      + " in the WITH cluase, KSQL used the default value. After version 5.1.x, KSQL will use the "


typo: cluase

big-andy-coates

Thanks @hjafarpour,

Please see my comment above about dropping this new LEGACY config you've added in favour of moving the two existing configs into 'ComatibilityBreakingConfigDef's.

Also, can you add a json based test that proves things work as expected when these legacy properties are, and are not, set please?

…versions using CompatibilityBreakingConfig.

big-andy-coates

Thanks @hjafarpour, this is looking good.

Usual nits below:

big-andy-coates · 2019-02-08T12:38:18Z

ksql-engine/src/test/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNodeTest.java

@@ -361,7 +362,10 @@ public void shouldThrowIfSinkTopicHasDifferentPropertiesThanRequested() {
  @Test
  public void shouldUseLegacySinkPropertiesIfLegacyIsTrue() {
    // Given:
-    when(ksqlConfig.getBoolean(KsqlConfig.KSQL_SINK_TOPIC_PROPERTIES_LEGACY_CONFIG)).thenReturn(true);
+    Mockito.<Object>when(ksqlConfig.values()).thenReturn((Map<String, ?>) ImmutableMap.of(


nit: drop the Mockito.<Object> and use the ImmutableMap.of generic args rather than casting:

Suggested change

Mockito.<Object>when(ksqlConfig.values()).thenReturn((Map<String, ?>) ImmutableMap.of(

when(ksqlConfig.values()).thenReturn(ImmutableMap.<Map<String, Object>of(

Removed the casting but still needed Mockito.<Object>!

big-andy-coates · 2019-02-08T12:44:48Z

ksql-engine/src/main/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNode.java

@@ -256,7 +256,11 @@ private static SourceTopicProperties getSourceTopicProperties(
      final KafkaTopicClient kafkaTopicClient,
      final KsqlConfig ksqlConfig
  ) {
-    if (ksqlConfig.getBoolean(KsqlConfig.KSQL_SINK_TOPIC_PROPERTIES_LEGACY_CONFIG)) {
+    final Map ksqlProperties = ksqlConfig.values();
+    if ((ksqlProperties.containsKey(KsqlConfig.SINK_NUMBER_OF_PARTITIONS_PROPERTY)


Can this not just be:

if (ksqlConfig.get(SINK_NUMBER_OF_PARTITIONS_PROPERTY) != null || ksqlConfig.get(SINK_NUMBER_OF_REPLICAS_PROPERTY) != null) { ... }

???

I don't think the containsKey adds anything as you're checking its not null and get returns null for any value that is not present.

Yes, removed the contains check.

big-andy-coates · 2019-02-08T12:53:41Z

ksql-engine/src/test/resources/query-validation-tests/sink-partitions-replicas.json

+{
+  "tests": [
+    {
+      "name": "Set correct sink partitions from the source topic",


nit: ambiguous name - it doesn't really explain what its testing ;)

Also, given this test isn't really about the inputs and outputs , it might benefit from a description field explaining how its testing this.

"name": "should copy partition and replica count from source topic", "description": "note: this isn't about input and outputs. It would fail if the counts are not copied across because the output topic 'S' already exists with the correct counts, and the execution path checks this"

Same for the other tests in this file. Alternatively, add this explanation once at the topic of the file.

{ "description": "as above", "tests" : [

Updated the names.

big-andy-coates · 2019-02-08T12:56:06Z

ksql-engine/src/test/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNodeTest.java

-    when(ksqlConfig.getBoolean(KsqlConfig.KSQL_SINK_TOPIC_PROPERTIES_LEGACY_CONFIG)).thenReturn(true);
+    Mockito.<Object>when(ksqlConfig.values()).thenReturn((Map<String, ?>) ImmutableMap.of(
+        KsqlConfig.SINK_NUMBER_OF_PARTITIONS_PROPERTY, KsqlConstants.defaultSinkNumberOfPartitions,
+        KsqlConfig.SINK_NUMBER_OF_REPLICAS_PROPERTY, KsqlConstants.defaultSinkNumberOfReplications


Probably worth splitting this into two tests, where each test only has one of these two properties set. This then tests the cases where the other is null.

Splited it into two tests, however, we cannot have one of them null and the other legacy. If we use legacy both should be null.

big-andy-coates · 2019-02-08T12:57:07Z

ksql-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java

+              Type.INT,
+              4,
+              null,
+              Importance.MEDIUM,


nit:

Suggested change

Importance.MEDIUM,

Importance.LOW,

big-andy-coates · 2019-02-08T13:13:35Z

ksql-engine/src/test/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNodeTest.java

+
+    // Then:
+    verify(mockTopicClient).createTopic(SINK_KAFKA_TOPIC_NAME, 1, (short) 2, Collections.emptyMap());
+    assertThat(schemaKStream, instanceOf(SchemaKStream.class));


pointless assert: schemaKStream is declared as an instance of SchemaKStream.

big-andy-coates · 2019-02-08T13:15:12Z

ksql-engine/src/test/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNodeTest.java

+    // Given:
+    doThrow(KsqlException.class).when(mockTopicClient).createTopic(SINK_KAFKA_TOPIC_NAME, 1, (short) 2, Collections.emptyMap());
+    createOutputNode(Collections.singletonMap(KsqlConfig.SINK_NUMBER_OF_REPLICAS_PROPERTY, (short) 2), true);
+    expectedException.expect(KsqlException.class);


super nit: Add a //Then on the line above the `expectedException. e.g. a negative test might look like:

// Given ... set up preconditions // Then: ... set up expected post conditions, i.e. the expected exception. // When: ... the line on test.

big-andy-coates · 2019-02-08T13:17:04Z

ksql-engine/src/test/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNodeTest.java

+        KsqlConfig.SINK_NUMBER_OF_PARTITIONS_PROPERTY, KsqlConstants.defaultSinkNumberOfPartitions,
+        KsqlConfig.SINK_NUMBER_OF_REPLICAS_PROPERTY, KsqlConstants.defaultSinkNumberOfReplications
+    ));
+    when(ksqlConfig.getInt(KsqlConfig.SINK_NUMBER_OF_PARTITIONS_PROPERTY)).thenReturn(KsqlConstants.defaultSinkNumberOfPartitions);


weird double mocking of KsqlConfig.SINK_NUMBER_OF_PARTITIONS_PROPERTY in this test - but I think this will go away once you simplify the if statement in the prod code this is testing.

big-andy-coates · 2019-02-08T13:18:19Z

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

@@ -110,6 +114,8 @@ public void setUp() {
    statementExecutor = new StatementExecutor(ksqlConfig, ksqlEngine, statementParser);
    statementExecutorWithMocks
        = new StatementExecutor(ksqlConfig, mockEngine, mockParser);
+//    doReturn(new TopicDescription("test1", true, Collections.singletonList(mock(TopicPartitionInfo.class))))


Can you tidy up the commented out code please?

big-andy-coates · 2019-02-08T13:20:33Z

ksql-rest-app/src/test/java/io/confluent/ksql/rest/server/mock/MockKafkaTopicClient.java

@@ -54,6 +57,14 @@ public boolean isTopicExists(final String topic) {
    return Collections.emptyMap();
  }

+  @Override
+  public TopicDescription describeTopic(final String topicName) {


Rather than hard coding your test requirements into the generic MockKafkaTopicClient you might be better of switching your test to use FakeKafkaTopicClient, which allows you to set up preconditions, like topics.

big-andy-coates

Hey @hjafarpour - meant to approve! Please take a look at my comments above, as they still need addressing, but assuming you do, then this LGTM!

…Props-If-Not-Specified

hjafarpour requested a review from a team September 21, 2018 22:53

big-andy-coates reviewed Sep 24, 2018

View reviewed changes

ksql-engine/src/main/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNode.java Outdated Show resolved Hide resolved

big-andy-coates requested a review from a team September 24, 2018 17:06

rodesai reviewed Sep 25, 2018

View reviewed changes

ksql-engine/src/test/java/io/confluent/ksql/KsqlEngineTest.java Outdated Show resolved Hide resolved

rodesai reviewed Sep 25, 2018

View reviewed changes

ksql-engine/src/test/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNodeTest.java Outdated Show resolved Hide resolved

rodesai reviewed Sep 25, 2018

View reviewed changes

hjafarpour added 2 commits September 25, 2018 14:05

Use source partition count and replica count as default for sink topics.

d79f6a6

Review feedback.

e794ca2

hjafarpour force-pushed the Use-Source-Topic-Props-If-Not-Specified branch from ab9b4c7 to e794ca2 Compare September 25, 2018 21:05

Minor test fix.

d18c7e7

hjafarpour added 5 commits December 18, 2018 11:08

Merge remote-tracking branch 'upstream/master' into Use-Source-Topic-…

18914f7

…Props-If-Not-Specified

Some improvements.

2fc0d15

Merge remote-tracking branch 'upstream/master' into Use-Source-Topic-…

56b55f9

…Props-If-Not-Specified

Some format fixes.

81d7858

Minor format fixes.

74d36f5

hjafarpour requested review from big-andy-coates and a team December 20, 2018 00:41

big-andy-coates reviewed Dec 21, 2018

View reviewed changes

hjafarpour added 4 commits December 27, 2018 11:10

Merge remote-tracking branch 'upstream/master' into Use-Source-Topic-…

18e0386

…Props-If-Not-Specified

Applied review feedback

65efb50

Merge remote-tracking branch 'upstream/master' into Use-Source-Topic-…

c5a683a

…Props-If-Not-Specified

Merge remote-tracking branch 'upstream/master' into Use-Source-Topic-…

fa59b4d

…Props-If-Not-Specified

vcrfxia approved these changes Jan 8, 2019

View reviewed changes

hjafarpour added 2 commits January 15, 2019 15:57

Merge remote-tracking branch 'upstream/master' into Use-Source-Topic-…

d08e213

…Props-If-Not-Specified

Applied feedback from Victoria.

12b77e8

hjafarpour requested a review from a team January 16, 2019 04:38

big-andy-coates requested review from big-andy-coates and rodesai and removed request for a team January 22, 2019 15:48

Merge remote-tracking branch 'upstream/master' into Use-Source-Topic-…

1bec128

…Props-If-Not-Specified

big-andy-coates reviewed Jan 23, 2019

View reviewed changes

hjafarpour added 5 commits January 25, 2019 09:58

Added legacy support

a50bcf8

Added legacy support

c9ac170

Merge remote-tracking branch 'upstream/master' into Use-Source-Topic-…

43b0b11

…Props-If-Not-Specified

Merge remote-tracking branch 'upstream/master' into Use-Source-Topic-…

582e709

…Props-If-Not-Specified

Removed unnecessary line.

b8cdae4

hjafarpour commented Feb 5, 2019

View reviewed changes

ksql-engine/src/main/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNode.java Show resolved Hide resolved

ksql-engine/src/main/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNode.java Show resolved Hide resolved

hjafarpour requested a review from big-andy-coates February 5, 2019 23:53

apurvam reviewed Feb 6, 2019

View reviewed changes

big-andy-coates reviewed Feb 6, 2019

View reviewed changes

hjafarpour added 2 commits February 7, 2019 14:29

Removed the legacy config and handle sink topic properties for older …

d264529

…versions using CompatibilityBreakingConfig.

Fixed a style bug.

9fb21e5

big-andy-coates reviewed Feb 8, 2019

View reviewed changes

big-andy-coates approved these changes Feb 8, 2019

View reviewed changes

This was referenced Feb 13, 2019

Write standalone mode configs to a kafka topic and read/merge before … #1491

Merged

Process commands so that WITH clause is always resolved before executing it #2436

Merged

hjafarpour added 2 commits March 5, 2019 10:21

Merge remote-tracking branch 'upstream/master' into Use-Source-Topic-…

9817f99

…Props-If-Not-Specified

Applied final feedback from Andy.

213dc43

hjafarpour merged commit 3cf18af into confluentinc:master Mar 6, 2019

vcrfxia mentioned this pull request Mar 8, 2019

Docs and follow-up to sink partitions/replicas change #2539

Closed

hjafarpour mentioned this pull request Mar 12, 2019

Added KSQL_INTERNAL_TOPIC_REPLICAS_PROPERTY to set the replicas for t… #2555

Merged

2 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Use source partition count and replica count as default for sink topics. #1930

Use source partition count and replica count as default for sink topics. #1930

hjafarpour commented Sep 21, 2018

big-andy-coates left a comment

rodesai Sep 25, 2018

hjafarpour Sep 25, 2018

big-andy-coates commented Sep 26, 2018

big-andy-coates commented Oct 18, 2018

hjafarpour commented Dec 20, 2018

big-andy-coates left a comment

vcrfxia left a comment

vcrfxia Jan 8, 2019

hjafarpour commented Jan 16, 2019

big-andy-coates left a comment

hjafarpour left a comment

apurvam Feb 6, 2019

big-andy-coates Feb 6, 2019 •

edited

Loading

apurvam Feb 6, 2019

big-andy-coates left a comment •

edited

Loading

big-andy-coates left a comment

big-andy-coates Feb 8, 2019

hjafarpour Mar 5, 2019

big-andy-coates Feb 8, 2019

hjafarpour Mar 4, 2019

big-andy-coates Feb 8, 2019

hjafarpour Mar 5, 2019

big-andy-coates Feb 8, 2019

hjafarpour Mar 5, 2019

big-andy-coates Feb 8, 2019

hjafarpour Mar 4, 2019

big-andy-coates Feb 8, 2019

big-andy-coates Feb 8, 2019

big-andy-coates Feb 8, 2019

big-andy-coates Feb 8, 2019

big-andy-coates Feb 8, 2019

hjafarpour Mar 5, 2019

big-andy-coates left a comment

	Mockito.<Object>when(ksqlConfig.values()).thenReturn((Map<String, ?>) ImmutableMap.of(
	when(ksqlConfig.values()).thenReturn(ImmutableMap.<Map<String, Object>of(

Use source partition count and replica count as default for sink topics. #1930

Use source partition count and replica count as default for sink topics. #1930

Conversation

hjafarpour commented Sep 21, 2018

Description

Testing done

Reviewer checklist

big-andy-coates left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

big-andy-coates commented Sep 26, 2018

big-andy-coates commented Oct 18, 2018

hjafarpour commented Dec 20, 2018

big-andy-coates left a comment

Choose a reason for hiding this comment

vcrfxia left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hjafarpour commented Jan 16, 2019

big-andy-coates left a comment

Choose a reason for hiding this comment

hjafarpour left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

big-andy-coates Feb 6, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

big-andy-coates left a comment • edited Loading

Choose a reason for hiding this comment

big-andy-coates 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

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

big-andy-coates left a comment

Choose a reason for hiding this comment

big-andy-coates Feb 6, 2019 •

edited

Loading

big-andy-coates left a comment •

edited

Loading