Ksql 386 ksql functions non static #370

hjafarpour · 2017-10-13T03:06:53Z

This PR refactors the function meta data handling by using a Function registry.

…uery termination.

…intermediate-topics-for-terminated-queries

dguy · 2017-10-13T10:23:32Z

build-tools/pom.xml

@@ -20,6 +20,6 @@
    <modelVersion>4.0.0</modelVersion>
    <groupId>io.confluent</groupId>
    <artifactId>build-tools</artifactId>
-    <version>4.0.0-SNAPSHOT</version>
+    <version>4.1.0-SNAPSHOT</version>


why 4.1.0-SNAPSHOT? Shouldn't this branch remain on 4.0?

Is 4.0.1 using kafka 11.0.1 where the adminclient is fixed?

It should be 4.0.0. I think it was messed up when I synced while Alex was preparing the branches. Will fix it.

dguy · 2017-10-13T10:27:56Z

ksql-clickstream-demo/pom.xml

@@ -23,7 +23,7 @@
    <parent>
        <groupId>io.confluent.ksql</groupId>
        <artifactId>ksql-parent</artifactId>
-        <version>4.0.0-SNAPSHOT</version>
+        <version>4.1.0-SNAPSHOT</version>


Same as above, will fix it :)

dguy · 2017-10-13T10:30:22Z

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

@@ -48,6 +48,10 @@
  public static final String DEFAULT_SINK_WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION =
      "ksql.sink.window.change.log.additional.retention.default";

+  public static final String STREAM_INTERNAL_CHANGELOG_TOPIC_SUFFIX = "-changelog";


Do these need to be public? Looks like they could be package protected and don't need to be known by a user?

dguy · 2017-10-13T10:35:48Z

ksql-engine/src/main/java/io/confluent/ksql/KsqlContext.java

@@ -61,8 +64,8 @@ public KsqlContext() {
      streamsProperties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BOOTSTRAP_SERVER_OPTION_DEFAULT);
    }
    KsqlConfig ksqlConfig = new KsqlConfig(streamsProperties);
-
-    topicClient = new KafkaTopicClientImpl(ksqlConfig.getKsqlAdminClientConfigProps());
+    adminClient = AdminClient.create(ksqlConfig.getKsqlAdminClientConfigProps());


As it stands we can't test this class without starting up kafka etc. I suggest we change it so that we can unit test the class. We could add one or two static factory methods, i.e,
KsqlContext.create() and KsqlContext.create(Map<String, Object> properties) we than have a package private constructor KsqlContext(AdminClient, KafkaTopicClient, KsqlEngine)
We can then have unit tests of this class

Good point! I'll make the changes.

dguy · 2017-10-13T10:38:43Z

ksql-engine/src/main/java/io/confluent/ksql/KsqlEngine.java

-                                         final MetaStore tempMetaStore)
-          throws Exception {
-
+  public List<QueryMetadata> planQueries(final boolean createNewAppId, final List<Pair<String, Statement>> statementList, final Map<String, Object> overriddenProperties, final MetaStore tempMetaStore) throws Exception {


with so many params and a long line length it would be good to have 1 param per line. Makes the diffs easier to read

dguy · 2017-10-13T10:59:56Z

ksql-engine/src/main/java/io/confluent/ksql/structured/SqlPredicate.java

@@ -120,7 +123,7 @@ private Predicate getWindowedKeyPredicate() throws Exception {
    CodeGenRunner codeGenRunner = new CodeGenRunner();
    ExpressionMetadata
        expressionEvaluator =
-        codeGenRunner.buildCodeGenFromParseTree(filterExpression, schema);
+        codeGenRunner.buildCodeGenFromParseTree(filterExpression, schema, ksqlFunctionRegistry);


pass KSqlFunctionRegistry into constructor of CodeGenRunner ?

Good idea! Done :)

dguy · 2017-10-13T11:00:51Z

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

+import java.util.List;
+import java.util.Set;
+
+public class CleanUpUtil {


I don't see where this class is used and there are no tests for it

I will use it in the clean up task.

code shouldn't be added without tests

dguy · 2017-10-13T11:02:05Z

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


-  public KafkaTopicClientImpl(Map<String, Object> adminClientConfig) {
-    this.adminClientConfig = adminClientConfig;
+  public KafkaTopicClientImpl(AdminClient adminClient) {


nit: final

dguy · 2017-10-13T11:02:22Z

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

    } catch (InterruptedException | ExecutionException e) {
      throw new KafkaResponseGetFailedException("Failed to describe kafka topics", e);
    }
  }

+  public void deleteTopics(List<String> topicsToDelete) {


tests for this method?

Will add test when I finish the clean up task which this method is for.

dguy · 2017-10-13T11:04:35Z

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

+        hasDeleteErrors = true;
+      }
+    }
+    if (hasDeleteErrors) {


rather than having a flag it would be better to collect all the failures into a list and then throw an exception with all the topics that failed. Also, we shouldn't be throwing RuntimeException - at the very least it should be KsqlException, but perhaps a DeleteTopicsFailedException or something.

Good idea, Done!

bluemonk3y · 2017-10-13T11:01:49Z

ksql-engine/src/main/java/io/confluent/ksql/codegen/SqlToJavaVisitor.java

@@ -165,7 +172,7 @@ private static String formatQualifiedName(QualifiedName name) {
                                                     Boolean unmangleNames) {
      StringBuilder builder = new StringBuilder("(");
      String name = node.getName().getSuffix();
-      KsqlFunction ksqlFunction = KsqlFunctions.getFunction(name);
+      KsqlFunction ksqlFunction = ksqlFunctionRegistry.getFunction(name);


Is it possible to remove the ksql prefixing everywhere?

Yer, consider it done :)

bluemonk3y · 2017-10-13T14:50:06Z

ksql-engine/src/main/java/io/confluent/ksql/function/KsqlFunctionRegistry.java

@@ -46,12 +46,16 @@
 import java.util.List;
 import java.util.Map;

-public class KsqlFunctions {
+public class KsqlFunctionRegistry {


Can rename this to FunctionRegistry - we have an an explosion of KSQLEverything

bluemonk3y · 2017-10-13T14:53:58Z

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


 public class KafkaTopicClientImpl implements KafkaTopicClient {
  private static final Logger log = LoggerFactory.getLogger(KafkaTopicClient.class);
-  private final Map<String, Object> adminClientConfig;
+  private final AdminClient adminClient;


Is 4.0.1 using kafka 11.0.1 where the adminclient is fixed?

we use 4.0.0 and it has Kafka 1.0

Yes, the adminclient should be fixed there. I also reimported your changes to use singleton adminclinent

bluemonk3y · 2017-10-13T14:54:10Z

build-tools/pom.xml

@@ -20,6 +20,6 @@
    <modelVersion>4.0.0</modelVersion>
    <groupId>io.confluent</groupId>
    <artifactId>build-tools</artifactId>
-    <version>4.0.0-SNAPSHOT</version>
+    <version>4.1.0-SNAPSHOT</version>


Is 4.0.1 using kafka 11.0.1 where the adminclient is fixed?

bluemonk3y · 2017-10-13T14:56:25Z

ksql-engine/src/test/java/io/confluent/ksql/structured/SqlPredicateTest.java

@@ -69,12 +72,12 @@ private PlanNode buildLogicalPlan(String queryStr) {
    analyzer.process(statements.get(0), new AnalysisContext(null));
    AggregateAnalysis aggregateAnalysis = new AggregateAnalysis();
    AggregateAnalyzer aggregateAnalyzer = new AggregateAnalyzer(aggregateAnalysis,
-        analysis);
+        analysis, ksqlFunctionRegistry);


would make me happy to rename this to functionRegistry

Definitely :) It's done!

…ions-non-static

dguy

Thanks @hjafarpour, still no tests for CleanupUtil and KafkaTopicClient.deleteTopics

dguy · 2017-10-17T08:55:59Z

ksql-engine/src/main/java/io/confluent/ksql/KsqlContext.java

@@ -50,7 +63,7 @@ public KsqlContext() {
   *
   * @param streamsProperties
   */
-  public KsqlContext(Map<String, Object> streamsProperties) {
+  private KsqlContext(Map<String, Object> streamsProperties) {


put the logic in here into the create(Map) method and have it call the new constructor below.

dguy · 2017-10-17T08:56:08Z

ksql-engine/src/main/java/io/confluent/ksql/KsqlContext.java

+    return new KsqlContext(streamsProperties);
+  }
+
+  private KsqlContext() {


can remove this

dguy · 2017-10-17T08:56:32Z

ksql-engine/src/main/java/io/confluent/ksql/KsqlContext.java

+
+
+  public static KsqlContext create() {
+    return new KsqlContext();


delegation, i.e, create(Collections.emptyMap())

Good point, made the changes.

dguy · 2017-10-17T14:32:58Z

@hjafarpour this PR doesn't look right. You have commits that are unrelated, i.e., those from @aayars

hjafarpour · 2017-10-17T15:15:50Z

@dguy the commits by alex came from a merge I had to do with master.
Also the tests for CleanUp and delete topic will be part of the next PR that I am working on.

dguy · 2017-10-17T15:52:41Z

@hjafarpour why would you need to merge master into 4.0.x? It should be the other way around

…ions-non-static

dguy

LGTM

Alex Ayars and others added 10 commits October 10, 2017 17:02

Bump to common 4.1.0-SNAPSHOT

0aca9a7

Merge remote-tracking branch 'origin/4.0.x'

2c82044

Merge remote-tracking branch 'origin/4.0.x'

ed1ee1e

Merge remote-tracking branch 'upstream/master'

27a0e1f

Better clean up for queries. Internal topics will be deletted after q…

06b59a9

…uery termination.

Added singleton AdminClient back.

181fa50

Merge remote-tracking branch 'upstream/4.0.x' into KSQL-383-clean-up-…

8b299f3

…intermediate-topics-for-terminated-queries

Fixed the pom version.

bcec9eb

Pom version fix.

93634bf

Refactored Function registry.

a4c863c

dguy suggested changes Oct 13, 2017

View reviewed changes

bluemonk3y suggested changes Oct 13, 2017

View reviewed changes

Applied review feedback.

ac3b527

bluemonk3y approved these changes Oct 16, 2017

View reviewed changes

Merge remote-tracking branch 'upstream/4.0.x' into KSQL-386-KsqlFunct…

e03d4d1

…ions-non-static

dguy reviewed Oct 17, 2017

View reviewed changes

Minor modification in KSQLContext.

984e51a

Merge remote-tracking branch 'upstream/4.0.x' into KSQL-386-KsqlFunct…

c9b3f75

…ions-non-static

dguy approved these changes Oct 17, 2017

View reviewed changes

hjafarpour merged commit de6772e into confluentinc:4.0.x Oct 17, 2017



		public static KsqlContext create() {
		return new KsqlContext();

Ksql 386 ksql functions non static #370

Ksql 386 ksql functions non static #370

Conversation

hjafarpour commented Oct 13, 2017

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

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

dguy 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

dguy commented Oct 17, 2017

hjafarpour commented Oct 17, 2017 • edited Loading

dguy commented Oct 17, 2017

dguy left a comment

Choose a reason for hiding this comment

hjafarpour commented Oct 17, 2017 •

edited

Loading