Merge branch '0.4.x' into pr_merge_from_0_3_x_to_0_4_x

Author: kapilchhajer

Makefile

@@ -8,10 +8,10 @@ SHELL=/bin/bash -o pipefail
 
 check-dependency = $(if $(shell command -v $(1)),,$(error Make sure $(1) is installed))
 
-CP_VERSION ?= 5.5.0
+CP_VERSION ?= 6.0.1
 OPERATOR_VERSION ?= 0
 
-KAFKA_CONNECT_DATAGEN_VERSION ?= 0.3.2
+KAFKA_CONNECT_DATAGEN_VERSION ?= 0.4.0
 AGGREGATE_VERSION = $(KAFKA_CONNECT_DATAGEN_VERSION)-$(CP_VERSION)
 OPERATOR_AGGREGATE_VERSION = $(AGGREGATE_VERSION).$(OPERATOR_VERSION)
 

README.md

@@ -14,7 +14,7 @@
 # Versions
 
 There are multiple [released versions](https://github.com/confluentinc/kafka-connect-datagen/releases) of this connector, starting with `0.1.0`.
-The instructions below use version `0.3.2` as an example, but you can substitute any of the other released versions.
+The instructions below use version `0.4.0` as an example, but you can substitute any of the other released versions.
 In fact, unless specified otherwise, we recommend using the latest released version to get all of the features and bug fixes.
 
 # Usage
@@ -28,7 +28,7 @@ Using the [Confluent Hub Client](https://docs.confluent.io/current/connect/manag
 To install a specific release version you can run: 
 
 ```bash
-confluent-hub install confluentinc/kafka-connect-datagen:0.3.2
+confluent-hub install confluentinc/kafka-connect-datagen:0.4.0
 ```
 
 or to install the latest released version:
@@ -40,12 +40,12 @@ confluent-hub install confluentinc/kafka-connect-datagen:latest
 ### Build connector from latest code
 
 Alternatively, you may build and install the `kafka-connect-datagen` connector from latest code.
-Here we use `v0.3.2` to reference the git tag for the `0.3.2` version, but the same pattern works for all released versions.
+Here we use `v0.4.0` to reference the git tag for the `0.4.0` version, but the same pattern works for all released versions.
 
 ```bash
-git checkout v0.3.2
+git checkout v0.4.0
 mvn clean package
-confluent-hub install target/components/packages/confluentinc-kafka-connect-datagen-0.3.2.zip
+confluent-hub install target/components/packages/confluentinc-kafka-connect-datagen-0.4.0.zip
 ```
 
 ### Run connector in local install
@@ -70,7 +70,7 @@ You can create a Docker image packaged with the locally built source by running
 make build-docker-from-local CP_VERSION=5.5.0
 ```
 
-This will build the connector from source and create a local image with an aggregate version number.  The aggregate version number is the kafka-connect-datagen connector version number and the Confluent Platform version number separated with a `-`.   The local kafka-connect-datagen version number is defined in the `pom.xml` file, and the Confluent Platform version defined in the [Makefile](Makfile).  An example of the aggregate version number might be: `0.3.2-5.5.0`.
+This will build the connector from source and create a local image with an aggregate version number.  The aggregate version number is the kafka-connect-datagen connector version number and the Confluent Platform version number separated with a `-`.   The local kafka-connect-datagen version number is defined in the `pom.xml` file, and the Confluent Platform version defined in the [Makefile](Makfile).  An example of the aggregate version number might be: `0.4.0-6.0.1`.
 
 Alternatively, you can install the `kafka-connect-datagen` connector from [Confluent Hub](https://www.confluent.io/connector/kafka-connect-datagen/) into a Docker image by running:
 ```bash
@@ -80,9 +80,9 @@ make build-docker-from-released CP_VERSION=5.5.0
 The [Makefile](Makefile) contains some default variables that affect the version numbers of both the installed `kafka-connect-datagen` as well as the base Confluent Platform version.  The variables are located near the top of the [Makefile](Makefile) with the following names and current default values:
 
 ```bash
-CP_VERSION ?= 5.5.0
+CP_VERSION ?= 6.0.1
 
-KAFKA_CONNECT_DATAGEN_VERSION ?= 0.3.2
+KAFKA_CONNECT_DATAGEN_VERSION ?= 0.4.0
 ```
 These values can be overriden with variable declarations before the `make` command.  For example:
 
@@ -123,9 +123,10 @@ Parameter | Description | Default
 `kafka.topic` | Topic to write to | 
 `max.interval` | Max interval between messages (ms) | 500
 `iterations` | Number of messages to send from each task, or less than 1 for unlimited | -1
-`schema.filename` | Filename of schema to use
+`schema.string` | The literal JSON-encoded Avro schema to use. Cannot be set with `schema.filename` or `quickstart`
+`schema.filename` | Filename of schema to use. Cannot be set with `schema.string` or `quickstart`
 `schema.keyfield` | Name of field to use as the message key
-`quickstart` | Name of [quickstart](https://github.com/confluentinc/kafka-connect-datagen/tree/master/src/main/resources) to use
+`quickstart` | Name of [quickstart](https://github.com/confluentinc/kafka-connect-datagen/tree/master/src/main/resources) to use. Cannot be set with `schema.string` or `schema.filename`
 
 ## Sample configurations
 
@@ -140,7 +141,7 @@ For an example of using the the Protobuf converter with kafka-connect-datagen, s
 ## Use a bundled schema specification
 
 There are a few quickstart schema specifications bundled with `kafka-connect-datagen`, and they are listed in this [directory](https://github.com/confluentinc/kafka-connect-datagen/tree/master/src/main/resources).
-To use one of these bundled schema, refer to [this mapping](https://github.com/confluentinc/kafka-connect-datagen/blob/master/src/main/java/io/confluent/kafka/connect/datagen/DatagenTask.java#L66-L73) and in the configuration file, set the parameter `quickstart` to the associated name.
+To use one of these bundled schema, refer to [this mapping](https://github.com/confluentinc/kafka-connect-datagen/blob/master/src/main/java/io/confluent/kafka/connect/datagen/DatagenTask.java#L75-L86) and in the configuration file, set the parameter `quickstart` to the associated name.
 For example:
 
 ```bash
@@ -246,8 +247,8 @@ To release new versions of the Docker images to Dockerhub (https://hub.docker.co
 The [Makefile](Makefile) contains some default variables that affect the version numbers of both the installed `kafka-connect-datagen` as well as the base Confluent Platform version.  The variables are located near the top of the [Makefile](Makefile) with the following names and current default values:
 
 ```bash
-CP_VERSION ?= 5.5.0
-KAFKA_CONNECT_DATAGEN_VERSION ?= 0.3.2
+CP_VERSION ?= 6.0.1
+KAFKA_CONNECT_DATAGEN_VERSION ?= 0.4.0
 OPERATOR_VERSION ?= 0 # Operator is a 'rev' version appended at the end of the CP version, like so: 5.5.0.0
 ```
 

config/connector_pageviews_string.config

@@ -0,0 +1,13 @@
+{
+  "name": "datagen",
+  "config": {
+    "connector.class": "io.confluent.kafka.connect.datagen.DatagenConnector",
+    "kafka.topic": "test1",
+    "schema.string": "{\"type\": \"record\", \"namespace\": \"ksql\", \"name\": \"pageviews\", \"fields\": [{\"type\": {\"type\": \"long\", \"format_as_time\": \"unix_long\", \"arg.properties\": {\"iteration\": {\"start\": 1, \"step\": 10}}}, \"name\": \"viewtime\"}, {\"type\": {\"type\": \"string\", \"arg.properties\": {\"regex\": \"User_[1-9]{0,1}\"}}, \"name\": \"userid\"}, {\"type\": {\"type\": \"string\", \"arg.properties\": {\"regex\": \"Page_[1-9][0-9]?\"}}, \"name\": \"pageid\"}]}",
+    "schema.keyfield": "pageid",
+    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
+    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
+    "value.converter.schemas.enable": "false",
+    "tasks.max": "1"
+  }
+}

docs/pull_request_template.md

@@ -0,0 +1,28 @@
+## Problem
+
+
+## Solution
+
+
+<!--- Mark x in the box. -->
+##### Does this solution apply anywhere else?
+- [ ] yes
+- [ ] no
+
+##### If yes, where?
+
+
+## Test Strategy
+
+
+<!--- Mark x in the box for all that apply. -->
+##### Testing done:
+- [ ] Unit tests
+- [ ] Integration tests
+- [ ] System tests
+- [ ] Manual tests
+
+## Release Plan
+<!--- Describe the release plan for this feature. -->
+<!-- Are you backporting or merging to master? -->
+<!-- If you are reverting or rolling back, is it safe? --> 

pom.xml

@@ -25,16 +25,17 @@
 
     <groupId>io.confluent.kafka.connect</groupId>
     <artifactId>kafka-connect-datagen</artifactId>
-    <version>0.3.4-SNAPSHOT</version>
+    <version>0.4.2-SNAPSHOT</version>
     <packaging>jar</packaging>
 
 
     <properties>
         <connect-runtime-version>2.0.0</connect-runtime-version>
         <confluent.avro.generator.version>0.3.1</confluent.avro.generator.version>
         <junit.version>4.12</junit.version>
+        <guava.version>32.0.1-jre</guava.version>
         <avro.version>1.11.3</avro.version>
-        <licenses.version>5.4.0</licenses.version>
+        <jackson.version>2.15.2</jackson.version>
         <maven.release.plugin.version>2.5.3</maven.release.plugin.version>
     </properties>
 
@@ -60,7 +61,7 @@
         <connection>scm:git:git://github.com/confluentinc/kafka-connect-datagen.git</connection>
         <developerConnection>scm:git:[email protected]:confluentinc/kafka-connect-datagen.git</developerConnection>
         <url>https://github.com/confluentinc/kafka-connect-datagen</url>
-        <tag>0.3.x</tag>
+        <tag>0.4.x</tag>
     </scm>
 
     <developers>
@@ -86,6 +87,35 @@
         </pluginRepository>
     </pluginRepositories>
 
+
+    <!-- pin transitive dependencies for CVEs -->
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>com.google.guava</groupId>
+                <artifactId>guava</artifactId>
+                <version>${guava.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.httpcomponents</groupId>
+                <artifactId>httpclient</artifactId>
+                <version>${httpclient.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>com.fasterxml.jackson</groupId>
+                <artifactId>jackson-bom</artifactId>
+                <version>${jackson.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+            <dependency>
+                <groupId>org.xerial.snappy</groupId>
+                <artifactId>snappy-java</artifactId>
+                <version>1.1.10.3</version>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
     <dependencies>
         <dependency>
             <groupId>org.apache.kafka</groupId>
@@ -113,6 +143,7 @@
             <version>${junit.version}</version>
             <scope>test</scope>
         </dependency>
+
         <dependency>
             <groupId>io.confluent.avro</groupId>
             <artifactId>avro-random-generator</artifactId>

src/main/java/io/confluent/kafka/connect/datagen/DatagenConnector.java

@@ -16,13 +16,17 @@
 
 package io.confluent.kafka.connect.datagen;
 
+import com.google.common.annotations.VisibleForTesting;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import java.util.stream.Collectors;
+import org.apache.kafka.common.config.Config;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.ConfigValue;
 import org.apache.kafka.connect.connector.Task;
 import org.apache.kafka.connect.source.SourceConnector;
 import org.slf4j.Logger;
@@ -34,6 +38,10 @@ public class DatagenConnector extends SourceConnector {
   private DatagenConnectorConfig config;
   private Map<String, String> props;
 
+  @VisibleForTesting
+  static final String SCHEMA_SOURCE_ERR =
+      "Must set exactly one of " + String.join(", ", DatagenConnectorConfig.schemaSourceKeys());
+
   @Override
   public String version() {
     return VersionUtil.getVersion();
@@ -76,4 +84,27 @@ public void stop() {
   public ConfigDef config() {
     return DatagenConnectorConfig.conf();
   }
+
+  @Override
+  public Config validate(Map<String, String> connectorConfigs) {
+    Config config = super.validate(connectorConfigs);
+    validateSchemaSource(config);
+    return config;
+  }
+
+  private void validateSchemaSource(Config config) {
+    List<ConfigValue> schemaSources = config.configValues().stream()
+        .filter(v -> DatagenConnectorConfig.isExplicitlySetSchemaSource(v.name(), v.value()))
+        .collect(Collectors.toList());
+    if (schemaSources.size() > 1) {
+      for (ConfigValue v : schemaSources) {
+        v.addErrorMessage(SCHEMA_SOURCE_ERR);
+      }
+    }
+    if (schemaSources.size() == 0) {
+      config.configValues().stream()
+          .filter(v -> DatagenConnectorConfig.schemaSourceKeys().contains(v.name()))
+          .forEach(v -> v.addErrorMessage(SCHEMA_SOURCE_ERR));
+    }
+  }
 }

src/main/java/io/confluent/kafka/connect/datagen/DatagenConnectorConfig.java

@@ -16,6 +16,8 @@
 
 package io.confluent.kafka.connect.datagen;
 
+import com.google.common.collect.ImmutableList;
+import java.util.List;
 import java.util.Map;
 
 import org.apache.kafka.common.config.AbstractConfig;
@@ -32,6 +34,8 @@ public class DatagenConnectorConfig extends AbstractConfig {
   public static final String ITERATIONS_CONF = "iterations";
   private static final String ITERATIONS_DOC = "Number of messages to send from each task, "
       + "or less than 1 for unlimited";
+  public static final String SCHEMA_STRING_CONF = "schema.string";
+  private static final String SCHEMA_STRING_DOC = "The literal JSON-encoded Avro schema to use";
   public static final String SCHEMA_FILENAME_CONF = "schema.filename";
   private static final String SCHEMA_FILENAME_DOC = "Filename of schema to use";
   public static final String SCHEMA_KEYFIELD_CONF = "schema.keyfield";
@@ -56,6 +60,7 @@ public static ConfigDef conf() {
         .define(KAFKA_TOPIC_CONF, Type.STRING, Importance.HIGH, KAFKA_TOPIC_DOC)
         .define(MAXINTERVAL_CONF, Type.LONG, 500L, Importance.HIGH, MAXINTERVAL_DOC)
         .define(ITERATIONS_CONF, Type.INT, -1, Importance.HIGH, ITERATIONS_DOC)
+        .define(SCHEMA_STRING_CONF, Type.STRING, "", Importance.HIGH, SCHEMA_STRING_DOC)
         .define(SCHEMA_FILENAME_CONF, Type.STRING, "", Importance.HIGH, SCHEMA_FILENAME_DOC)
         .define(SCHEMA_KEYFIELD_CONF, Type.STRING, "", Importance.HIGH, SCHEMA_KEYFIELD_DOC)
         .define(QUICKSTART_CONF, Type.STRING, "", Importance.HIGH, QUICKSTART_DOC)
@@ -90,5 +95,16 @@ public Long getRandomSeed() {
     return this.getLong(RANDOM_SEED_CONF);
   }
 
+  public String getSchemaString() {
+    return this.getString(SCHEMA_STRING_CONF);
+  }
+
+  public static List<String> schemaSourceKeys() {
+    return ImmutableList.of(SCHEMA_STRING_CONF, SCHEMA_FILENAME_CONF, QUICKSTART_CONF);
+  }
+
+  public static boolean isExplicitlySetSchemaSource(String key, Object value) {
+    return schemaSourceKeys().contains(key) && !("".equals(value));
+  }
 }