Changed exception calls to use FailureCollector #67

vravish · 2019-09-19T04:59:54Z

Using the new Failure collectors to store several errors and not stop after the first one.

kafka-plugins-0.10/src/main/java/io/cdap/plugin/alertpublisher/KafkaAlertPublisher.java

albertshau · 2019-09-20T00:25:11Z

kafka-plugins-common/src/main/java/io/cdap/plugin/batch/source/KafkaBatchConfig.java

      if (fieldName.equals(keyField)) {
        if (fieldType != Schema.Type.BYTES) {
-          throw new IllegalArgumentException("The key field must be of type bytes or nullable bytes.");
+          collector.addFailure("The key field must be of type bytes or nullable bytes.", null);


.withConfigProperty("keyField") .withOutputSchemaField(keyField)

This should highlight both the keyField textbox as well as the corresponding field in the output schema.

albertshau · 2019-09-20T00:25:45Z

kafka-plugins-common/src/main/java/io/cdap/plugin/batch/source/KafkaBatchConfig.java

      } else if (fieldName.equals(partitionField)) {
        if (fieldType != Schema.Type.INT) {
-          throw new IllegalArgumentException("The partition field must be of type int.");
+          collector.addFailure("The partition field must be of type int.", null);


.withConfigProperty("partitionField") .withOutputSchemaField(partitionField)

albertshau · 2019-09-20T00:26:07Z

kafka-plugins-common/src/main/java/io/cdap/plugin/batch/source/KafkaBatchConfig.java

      } else if (fieldName.equals(offsetField)) {
        if (fieldType != Schema.Type.LONG) {
-          throw new IllegalArgumentException("The offset field must be of type long.");
+          collector.addFailure("The offset field must be of type long.", null);


.withConfigProperty("offsetField") .withOutputSchemaField(offsetField)

albertshau · 2019-09-20T00:27:48Z

kafka-plugins-common/src/main/java/io/cdap/plugin/batch/source/KafkaBatchConfig.java

-      throw new IllegalArgumentException(String.format(
-        "keyField '%s' does not exist in the schema. Please add it to the schema.", keyField));
+      collector.addFailure(String.format(
+        "keyField '%s' does not exist in the schema. Please add it to the schema.", keyField), null);


.withConfigProperty("keyField")

should also create a constant private static final String KEY_FIELD = "keyField"; and use that everywhere I'm writing "keyField"

albertshau · 2019-09-20T00:38:28Z

kafka-plugins-common/src/main/java/io/cdap/plugin/batch/source/KafkaBatchConfig.java

+        collector.addFailure(String.format(
          "Unable to instantiate a message parser from format '%s' and message schema '%s': %s",
-          format, messageSchema, e.getMessage()), e);
+          format, messageSchema, e.getMessage()), null);


.withConfigProperty("format")

albertshau · 2019-09-20T00:39:02Z

kafka-plugins-common/src/main/java/io/cdap/plugin/batch/source/KafkaBatchConfig.java

+        collector.addFailure(String.format(
          "Without a format, the schema must contain just a single message field of type bytes or nullable bytes. " +
-            "Found %s message fields (%s).", messageFields.size(), fieldNames));
+            "Found %s message fields (%s).", messageFields.size(), fieldNames), null);


.withConfigProperty("format")

kafka-plugins-0.10/src/main/java/io/cdap/plugin/alertpublisher/KafkaAlertPublisher.java

kafka-plugins-0.10/src/main/java/io/cdap/plugin/batch/source/KafkaBatchSource.java

albertshau · 2019-09-20T00:42:53Z

kafka-plugins-0.10/src/main/java/io/cdap/plugin/batch/source/KafkaBatchSource.java

    kafkaConf.putAll(config.getKafkaProperties());
    kafkaRequests = KafkaInputFormat.saveKafkaRequests(conf, config.getTopic(), kafkaConf,
-                                                       config.getPartitions(), config.getMaxNumberRecords(),
+                                                       config.getPartitions(context.getFailureCollector()), config.getMaxNumberRecords(),


should perform all these gets at the start of the method, then call getOrThrowException on the failure collector.

kafka-plugins-0.10/src/main/java/io/cdap/plugin/alertpublisher/KafkaAlertPublisher.java

albertshau · 2019-09-20T16:24:22Z

kafka-plugins-0.10/src/main/java/io/cdap/plugin/batch/source/KafkaBatchSource.java

-    config.validate();
-    pipelineConfigurer.getStageConfigurer().setOutputSchema(config.getSchema());
+    config.validate(pipelineConfigurer.getStageConfigurer().getFailureCollector());
+    pipelineConfigurer.getStageConfigurer().setOutputSchema(config.getSchema(pipelineConfigurer.getStageConfigurer()


nit: this is pretty verbose, let's create a variable for the StageConfigurer so that it's easier to read.

albertshau · 2019-09-20T16:26:57Z

kafka-plugins-0.10/src/main/java/io/cdap/plugin/batch/source/KafkaBatchSource.java

    kafkaConf.putAll(config.getKafkaProperties());
    kafkaRequests = KafkaInputFormat.saveKafkaRequests(conf, config.getTopic(), kafkaConf,
-                                                       config.getPartitions(), config.getMaxNumberRecords(),
+                                                       config.getPartitions(context.getFailureCollector()),


move this to the start of the method so that all the validation happens once at the start and so that execution is stopped if anything is invalid. It should look something like:

FailureCollector failureCollector = context.getFailureCollector(); KafkaPartitionOffsets partitionOffsets = config.getInitialPartitionOffsets(failureCollector); Partitions partitions = config.getPartitions(failureCollector); Schema schema = config.getSchema(failureCollector); failureCollector.getOrThrowException(); ...

albertshau · 2019-09-20T16:27:57Z

kafka-plugins-0.10/src/main/java/io/cdap/plugin/common/KafkaHelpers.java

+   * @param collector input failureCollector into which the error will be added if present
+   */
+  public static void validateKerberosSetting(@Nullable String principal, @Nullable String keytab,
+      FailureCollector collector) {


nit: fix indentation

albertshau · 2019-09-20T16:29:21Z

kafka-plugins-0.10/src/main/java/io/cdap/plugin/sink/KafkaBatchSink.java

+    private void validate(FailureCollector collector) {
      if (!async.equalsIgnoreCase("true") && !async.equalsIgnoreCase("false")) {
-        throw new IllegalArgumentException("Async flag has to be either TRUE or FALSE.");
+        collector.addFailure("Async flag has to be either TRUE or FALSE.", null);


withConfigProperty(ASYNC)

Use constants instead of hardcoded strings everywhere. This comment applies to the other classes as well. For example:

public static class Config extends ReferencePluginConfig { public static final String ASYNC = "async"; @Name(ASYNC) @Description("Specifies whether an acknowledgment is required from broker that message was received. " + "Default is FALSE") @Macro private String async; ... }

albertshau · 2019-09-20T17:02:05Z

kafka-plugins-common/src/main/java/io/cdap/plugin/batch/source/KafkaBatchConfig.java

-          String.format("Invalid partition '%s'. Partitions must be integers.", partition));
+        collector.addFailure(
+          String.format("Invalid partition '%s'. Partitions must be integers.", partition), null)
+                .withConfigElement("partitions", partition);


indentation

albertshau · 2019-09-20T17:03:43Z

kafka-plugins-common/src/main/java/io/cdap/plugin/batch/source/KafkaBatchConfig.java

-  public Schema getMessageSchema() {
-    Schema schema = getSchema();
+  public Schema getMessageSchema(FailureCollector collector) {
+    Schema schema = getSchema(collector);


if this has a problem, schema will be null and you'll get a NullPointerException below. Need to do a null check to avoid that.

albertshau · 2019-09-20T17:04:32Z

kafka-plugins-common/src/main/java/io/cdap/plugin/batch/source/KafkaBatchConfig.java

-      throw new IllegalArgumentException(
-        "Schema must contain at least one other field besides the time and key fields.");
+      collector.addFailure("Schema must contain at least one other field besides the time and key fields.", null)
+              .withConfigProperty("schema");


indentation. Similar comment throughout the PR.

albertshau · 2019-09-20T17:06:41Z

kafka-plugins-0.10/src/main/java/io/cdap/plugin/alertpublisher/KafkaAlertPublisher.java

-        throw new IllegalArgumentException(String.format("Topic name %s is not a valid kafka topic. Please provide " +
-                                                           "valid kafka topic name. %s", topic, e.getMessage()));
+        collector.addFailure(String.format("Topic name %s is not a valid kafka topic. Please provide " +
+                                                           "valid kafka topic name. %s", topic, e.getMessage()), null)


indentation

albertshau · 2019-09-20T17:09:58Z

kafka-plugins-0.10/src/main/java/io/cdap/plugin/common/KafkaHelpers.java

+      String message = emptyField + " is empty. When Kerberos security is enabled for Kafka, " +
+          "then both the principal and the keytab have " +
+          "to be specified. If Kerberos is not enabled, then both should be empty.";
+      collector.addFailure(message, null);


this needs to add a cause so that whatever field(s) have problems are highlighted. Similarly, the message should be specific to what is wrong.

In other words, if the principal is defined and the keytab is not, the keytab should be added as a cause, with a simple message saying the keytab must be specified. If the keytab is defined and the principal is not, the principal should be added as a cause with a message saying the principal must be specified.

albertshau · 2019-09-23T05:35:55Z

kafka-plugins-0.10/src/main/java/io/cdap/plugin/alertpublisher/KafkaAlertPublisher.java

-        throw new IllegalArgumentException(String.format("Topic name %s is not a valid kafka topic. Please provide " +
-                                                           "valid kafka topic name. %s", topic, e.getMessage()));
+        collector.addFailure(String.format(
+                "Topic name %s is not a valid kafka topic. Please provide valid kafka topic name. %s", topic,


indentation is still off in this PR when there is a new line. This should either line up with the arguments from the previous line or be indented 2 spaces if its a method call.

collector.addFailure(String.format( "Topic name %s is not a valid kafka topic. Please provide valid kafka topic name. %s", topic, e.getMessage()), null) .withConfigProperty(TOPIC);

albertshau · 2019-09-23T05:37:16Z

kafka-plugins-0.10/src/main/java/io/cdap/plugin/batch/source/KafkaBatchSource.java


-    KafkaPartitionOffsets partitionOffsets = config.getInitialPartitionOffsets();
+    FailureCollector failureCollector = context.getFailureCollector();
+    KafkaPartitionOffsets partitionOffsets = config.getInitialPartitionOffsets(failureCollector);


as mentioned before, do all the validation here before anything else happens. This should be:

FailureCollector failureCollector = context.getFailureCollector(); KafkaPartitionOffsets partitionOffsets = config.getInitialPartitionOffsets(failureCollector); Schema schema = config.getSchema(failureCollector); ... partitions = config.getPartitions(failureCollector); failureCollector.getOrThrowException();

Ok, but I previously moved FailureCollector later in the function, since I didn't think the if block or the saveKafkaRequests call needed to be done after getOrThrowException.

albertshau · 2019-09-23T05:38:18Z

kafka-plugins-0.10/src/main/java/io/cdap/plugin/batch/source/KafkaBatchSource.java

    kafkaConf.putAll(config.getKafkaProperties());
    kafkaRequests = KafkaInputFormat.saveKafkaRequests(conf, config.getTopic(), kafkaConf,
-                                                       config.getPartitions(), config.getMaxNumberRecords(),
+                                                       config.getPartitions(failureCollector),


this should be a variable that was calculated at the start of the method.

Before or after the call to getOrThrowException? And which variable are you referring to?

Never mind, I just understood that you wanted me to move every reference to failureCollector to the beginning (including getPartitions), right before the call to getOrThrowException

albertshau · 2019-09-23T05:38:22Z

kafka-plugins-0.10/src/main/java/io/cdap/plugin/batch/source/KafkaBatchSource.java

                                                       partitionOffsets);
    LineageRecorder lineageRecorder = new LineageRecorder(context, config.referenceName);
-    Schema schema = config.getSchema();
+    Schema schema = config.getSchema(failureCollector);


this should be a variable that was calculated at the start of the method.

albertshau · 2019-09-23T05:38:49Z

kafka-plugins-0.10/src/main/java/io/cdap/plugin/batch/source/KafkaBatchSource.java

    LineageRecorder lineageRecorder = new LineageRecorder(context, config.referenceName);
-    Schema schema = config.getSchema();
+    Schema schema = config.getSchema(failureCollector);
+    failureCollector.getOrThrowException();


this should be at the top, as mentioned above

albertshau · 2019-09-23T05:41:37Z

kafka-plugins-0.10/src/main/java/io/cdap/plugin/common/KafkaHelpers.java

+                                             FailureCollector collector) {
+    if (Strings.isNullOrEmpty(principal) != Strings.isNullOrEmpty(keytab)) {
+      String emptyField = Strings.isNullOrEmpty(principal) ? "principal" : "keytab";
+      String message = emptyField + " is empty. When Kerberos security is enabled for Kafka, then both the principal " +


If the principal is defined but the keytab is not, this should be a simple message: "A keytab must be specified."

If the keytab is defined by the principal is not, there should be a simple message: "A principal must be specified."

No need for a corrective action.

albertshau · 2019-09-23T05:42:06Z

kafka-plugins-0.10/src/main/java/io/cdap/plugin/common/KafkaHelpers.java

+  public static void validateKerberosSetting(@Nullable String principal, @Nullable String keytab,
+                                             FailureCollector collector) {
+    if (Strings.isNullOrEmpty(principal) != Strings.isNullOrEmpty(keytab)) {
+      String emptyField = Strings.isNullOrEmpty(principal) ? "principal" : "keytab";


principal and keytab should also be constants, not hardcoded strings

albertshau

one minor comment. Please fix then squash commits and we'll merge. Thanks for your contribution!

albertshau · 2019-09-25T03:35:04Z

kafka-plugins-0.8/src/main/java/io/cdap/plugin/alertpublisher/KafkaAlertPublisher.java

      } catch (InvalidTopicException e) {
-        throw new IllegalArgumentException(String.format("Topic name %s is not a valid kafka topic. Please provide " +
-                                                           "valid kafka topic name. %s", topic, e.getMessage()));
+        collector.addFailure(String.format("Topic name %s is not a valid kafka topic. Please provide valid kafka" +


missing space between 'kafka' and 'topic' on the next line

albertshau · 2019-09-25T22:23:14Z

due to timing, fixed indentation and squashed commits.

vravish requested a review from albertshau September 19, 2019 04:59

albertshau reviewed Sep 20, 2019

View reviewed changes

albertshau reviewed Sep 23, 2019

View reviewed changes

albertshau reviewed Sep 25, 2019

View reviewed changes

Changed exception calls to use FailureCollector

58641ed

albertshau force-pushed the feature_release/add-validation branch from 91af377 to 58641ed Compare September 25, 2019 22:22

albertshau merged commit 9d0afe2 into release/2.2 Sep 25, 2019

albertshau deleted the feature_release/add-validation branch September 25, 2019 22:23

Changed exception calls to use FailureCollector #67

Changed exception calls to use FailureCollector #67

Uh oh!

Conversation

vravish commented Sep 19, 2019

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vravish Sep 24, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

albertshau left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

albertshau commented Sep 25, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

vravish Sep 24, 2019 •

edited

Loading