Set the clean up policy to compact for non-windowed tables. #1042

hjafarpour · 2018-03-24T05:03:57Z

Currently we don't set the clean up policy for sink topics and it uses the default which is delete. For the non-window tables we should use compact since these are changelog topics.
For the windowed tables we still use delete clean up policy since we would prefer to not keep the windowed table forever!

apurvam

Looking good.

My main comment is that we need to add unit test cases in KSQLStructuredDataOutputNodeTest which verifies that non windowed tables actually are created with compaction enabled, that windowed ktables are not created with compaction, and that kstreams are not created with compaction.

apurvam · 2018-03-24T17:13:53Z

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

+                    ksqlConfig,
+                    kafkaTopicClient,
+                    ((result instanceof SchemaKTable)
+                     && !((SchemaKTable) result).isWindowed()));


I would move this boolean to a helper method named shouldCompact. Makes the intent much clearer.

apurvam · 2018-03-24T17:15:22Z

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

@@ -129,6 +137,12 @@ public boolean isTopicExists(final String topic) {
    }
  }

+  @Override
+  public DescribeConfigsResult describeConfigs(String topicName) {
+    return adminClient.describeConfigs(


This should use the RetryHelper.executeWithRetries method. Otherwise the command will fail for even retriable exceptions.

Changed the method to use RetryHelper and return the clean up policy only.

apurvam · 2018-03-24T17:17:50Z

ksql-engine/src/test/java/io/confluent/ksql/integration/JsonFormatTest.java

@@ -187,6 +203,44 @@ public void testSinkProperties() throws Exception {
    assertThat(
        topicClient.describeTopics(ImmutableList.of(streamName)).get(streamName).partitions(),
        hasSize(3));
+    Map<ConfigResource, Config> configResourceConfigMap = topicClient.describeConfigs


This block is used 3 times. Should be moved to a utility method named verifyCleanupPolicy(String policy). Then you can get the property from the cluster and verify that it matches the expectation.

With the topic client api change this will be one line only.

apurvam · 2018-03-24T17:18:57Z

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

@@ -92,7 +93,7 @@ public void before() {
    props.put(KsqlConfig.SINK_NUMBER_OF_PARTITIONS_PROPERTY, 4);
    props.put(KsqlConfig.SINK_NUMBER_OF_REPLICAS_PROPERTY, (short)3);
    createOutputNode(props);
-    topicClient.createTopic(eq("output"), anyInt(), anyShort());
+    topicClient.createTopic(eq("output"), anyInt(), anyShort(), anyBoolean());


We should add a unit test here that verifies that non windowed KTables actually are being created with cleanup.polic=compact.

Added a unit test.

rodesai

Changes look good. Left some minor comments inline

rodesai · 2018-03-24T18:57:21Z

ksql-engine/src/test/java/io/confluent/ksql/integration/JsonFormatTest.java

+  }
+
+  @Test
+  public void testSinkCleanupPolicy() throws Exception {


nit: rename to testTableSinkCleanupProperty

rodesai · 2018-03-24T18:59:26Z

ksql-engine/src/test/java/io/confluent/ksql/integration/WindowingIntTest.java

+            .filter
+                (configEntry -> configEntry.name().equalsIgnoreCase("cleanup.policy")).toArray();
+    assertThat(configEntries[0], instanceOf(ConfigEntry.class));
+    assertThat(((ConfigEntry) configEntries[0]).value(), equalTo("compact"));


Should we add a similar check to shouldAggregateTumblingWindow that verifies that the policy is "delete"?

rodesai · 2018-03-24T19:11:00Z

ksql-engine/src/test/java/io/confluent/ksql/util/KafkaTopicClientTest.java

    DescribeConfigsResult describeConfigsResult = mock(DescribeConfigsResult.class);
-    ConfigEntry configEntry = new ConfigEntry("delete.topic.enable", "true");
+    ConfigEntry configEntryDeleteEnable = new ConfigEntry("delete.topic.enable", "true");
+    ConfigEntry configEntryIsCompacted = new ConfigEntry("cleanup.policy",


Is this returned config entry actually used anywhere in admin client? The only usage of describeConfigs I see is to get the broker's delete.topic.enable config. If cleanup.policy is not read anywhere can we just get rid of it and the isCompacted parameter to this function.

No it was only used in tests. Removed it.

…-retention-policy

rodesai · 2018-03-26T23:26:29Z

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

@@ -74,11 +73,12 @@ void createTopic(
  Map<String, TopicDescription> describeTopics(Collection<String> topicNames);

  /**
-   * [warn] synchronous call to get the response
+   * [warn] Asynchronous call to get the response


this call looks synchronous to me

rodesai · 2018-03-26T23:29:38Z

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

+                                                                      new FunctionRegistry(),
+                                                                      new HashMap<>(),
+                                                                      new MockSchemaRegistryClient());
+    topicClientForNonWindowTable.createTopic(eq("output"), anyInt(), anyShort(), eq(true));


Shouldn't this plus the next 2 lines come before buildStream?

Yes, indeed this needed more changes.

apurvam · 2018-03-26T23:07:13Z

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

@@ -74,11 +73,12 @@ void createTopic(
  Map<String, TopicDescription> describeTopics(Collection<String> topicNames);

  /**
-   * [warn] synchronous call to get the response
+   * [warn] Asynchronous call to get the response


This is a synchronous call. It blocks until it gets a successful response or will throw an exception if it didn't get a successful response after 5 retries.

apurvam · 2018-03-26T23:07:29Z

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

   *
-   * @param topicName topicName to describe
+   * @param topicName topicNames to describe


Why make this plural. It is still a single topic.

This calls AdminClient.describeTopics which gets a collection of topics.

Yes. But this method just takes a single topic and passes a list with a single element and also assumes the output contains a single pair.

So it is more appropriate for javadoc for this method to be singular rather than plural.

apurvam · 2018-03-26T23:09:04Z

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

   */
-  DescribeConfigsResult describeConfigs(String topicName);
+  public String getTopicCleanupPolicy(String topicName);


Would be better if this was an enum with three values: delete, compact, compact+delete. Would be better than each user of the method relying on the actual string literals.

apurvam · 2018-03-26T23:27:19Z

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

+                                                  Optional.empty());
+
+    StreamsBuilder streamsBuilder = new StreamsBuilder();
+    SchemaKTable schemaKTable = (SchemaKTable) outputNode.buildStream(streamsBuilder,


We should first have a check assertThat(scheamKstream, instanceof(schemaKTable)) and then do the cast.

apurvam · 2018-03-26T23:27:49Z

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

+                                                                      new FunctionRegistry(),
+                                                                      new HashMap<>(),
+                                                                      new MockSchemaRegistryClient());
+    topicClientForNonWindowTable.createTopic(eq("output"), anyInt(), anyShort(), eq(true));


Would be better to have 4 and3 instead fo anyInt() and anyShort().

apurvam · 2018-03-26T23:28:33Z

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

+                                                                      new MockSchemaRegistryClient());
+    topicClientForNonWindowTable.createTopic(eq("output"), anyInt(), anyShort(), eq(true));
+    EasyMock.expectLastCall();
+    EasyMock.replay(topicClientForNonWindowTable);


This seems off.

We should be doing:

topicClientForNonWindowTable.createTopic(...) Easymock.expectLastCall(); EasyMock.replay(topicClientForNonWindownTable); outputNode.buildStream(...); EasyMock.verify(topicClientForNonWindowTable);

This way we are actually checking that the right call was made to the topic client when trying to create the sink topic.

Also, we should add tests which check that the compaction policy is delete for streams and windowed tables.

Made changes and added tests for both windowed table and stream.

apurvam

Almost there. My main comment is that we should start using the values of the new enums in the tests. In fact, I am not sure how the code compiled.

apurvam · 2018-03-27T00:38:56Z

ksql-engine/src/test/java/io/confluent/ksql/integration/JsonFormatTest.java

+        "Wait for async topic creation"
+    );
+
+    assertThat(topicClient.getTopicCleanupPolicy(tableName), equalTo("compact"));


here and elsewhere we should be checking equivalence to TopicCleanupPolicy.COMPACT etc.

apurvam · 2018-03-27T00:39:20Z

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

+    topicClientForNonWindowTable.createTopic("output", 4, (short) 3, true);
+    EasyMock.replay(topicClientForNonWindowTable);
+    SchemaKStream schemaKStream = outputNode.buildStream(streamsBuilder,
+                                                                      ksqlConfig,


indentation.

apurvam · 2018-03-27T00:39:28Z

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

+        schema);
+
+    return new KsqlStructuredDataOutputNode(new PlanNodeId("0"),
+                                                  tableSourceNode,


indentation.

rodesai · 2018-03-27T01:13:36Z

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

@@ -64,24 +66,31 @@ public KafkaTopicClientImpl(final AdminClient adminClient) {
  public void createTopic(
      final String topic,
      final int numPartitions,
-      final short replicatonFactor
+      final short replicatonFactor,
+      boolean isCompacted


We should use the TopicCleanupPolicy here instead of isCompacted. Also, wouldn't it be better to use compact+delete for topics backing windowed aggregations?

Both of these changes can be made in a later PR targeting master

apurvam

Looking good. We should cleanup the KafkaTopicClientTest before merging: some of those changes seem redundant and were perhaps added during an early iteration.

apurvam · 2018-03-27T01:12:30Z

ksql-engine/src/test/java/io/confluent/ksql/util/KafkaTopicClientTest.java

@@ -316,12 +317,13 @@ private DescribeClusterResult getDescribeClusterResult() {
    return describeClusterResult;
  }

-  private DescribeConfigsResult getDescribeConfigsResult() {
+  private DescribeConfigsResult getDescribeConfigsResult(boolean isCompacted) {


Why this change? This variable doesn't seem to be used at all and really doesn't need to be changed relative to the other changes in this PR.

Yes, it was left from previous changes. Removed it.

rodesai

lgtm modulo one nit

rodesai · 2018-03-27T03:56:25Z

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

+        .stream()
+        .filter(configEntry -> configEntry.name().equalsIgnoreCase("cleanup.policy"))
+        .toArray();
+    if (configValues == null || configValues.length ==0) {


Set the clean up policy to compact for non-windowed tables.

91c2193

hjafarpour added the bug label Mar 24, 2018

hjafarpour self-assigned this Mar 24, 2018

hjafarpour requested a review from a team March 24, 2018 05:03

apurvam reviewed Mar 24, 2018

View reviewed changes

rodesai reviewed Mar 24, 2018

View reviewed changes

hjafarpour added 2 commits March 26, 2018 13:36

Merge remote-tracking branch 'upstream/4.1.x' into KSQL-1038-Set-sink…

fcaff17

…-retention-policy

Applied review feedback

c7e94f0

rodesai reviewed Mar 26, 2018

View reviewed changes

apurvam reviewed Mar 26, 2018

View reviewed changes

More reveiw feedback.

1273ba2

apurvam reviewed Mar 27, 2018

View reviewed changes

Style changes and review feedbackStyle changes and review feedback..

915c19e

rodesai reviewed Mar 27, 2018

View reviewed changes

apurvam approved these changes Mar 27, 2018

View reviewed changes

rodesai approved these changes Mar 27, 2018

View reviewed changes

More review feedback.

226a837

hjafarpour merged commit 1dd483f into confluentinc:4.1.x Mar 27, 2018

This was referenced Mar 27, 2018

Set the sink topic retention policy correctly #1038

Closed

Add test to verify that cleanup.policy is set to compact in the KafkaTopicClient #1071

Merged

agavra mentioned this pull request Jun 4, 2020

Windowed Tables should use [delete, compact] for their Sink Topics #5554

Closed

Set the clean up policy to compact for non-windowed tables. #1042

Set the clean up policy to compact for non-windowed tables. #1042

Conversation

hjafarpour commented Mar 24, 2018

apurvam 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

rodesai 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

apurvam Mar 26, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apurvam 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

apurvam left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rodesai left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apurvam Mar 26, 2018 •

edited

Loading