Added basic authentication to all requests. Fixes #17.

src/main/java/com/github/jcustenborder/kafka/connect/solr/CloudSolrSinkConnector.java

@@ -16,11 +16,17 @@
 package com.github.jcustenborder.kafka.connect.solr;
 
 import com.github.jcustenborder.kafka.connect.utils.config.Description;
+import com.github.jcustenborder.kafka.connect.utils.config.DocumentationTip;
+import com.github.jcustenborder.kafka.connect.utils.config.Title;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.connect.connector.Task;
 
+@Title("Cloud Solr")
 @Description("This connector is used to connect to `SolrCloud <https://cwiki.apache.org/confluence/display/solr/SolrCloud>`_ " +
     "using the Zookeeper based configuration.")
+@DocumentationTip("The target collection for this connector is selected by the topic name. " +
+    "`Transformations <https://kafka.apache.org/documentation/#connect_transforms>`_ like the " +
+    "RegexRouter transformation can be used to change the topic name before it is sent to Solr.")
 public class CloudSolrSinkConnector extends SolrSinkConnector {
   @Override
   public Class<? extends Task> taskClass() {

src/main/java/com/github/jcustenborder/kafka/connect/solr/CloudSolrSinkConnectorConfig.java

@@ -15,6 +15,7 @@
  */
 package com.github.jcustenborder.kafka.connect.solr;
 
+import com.github.jcustenborder.kafka.connect.utils.config.ConfigKeyBuilder;
 import org.apache.kafka.common.config.ConfigDef;
 
 import java.util.List;
@@ -27,24 +28,32 @@ class CloudSolrSinkConnectorConfig extends SolrSinkConnectorConfig {
   public static final String COLLECTION_NAME_CONFIG = "solr.collection.name";
   private static final String ZOOKEEPER_HOSTS_DOC = "Zookeeper hosts that are used to store solr configuration.";
   private static final String ZOOKEEPER_CHROOT_DOC = "Chroot within solr for the zookeeper configuration.";
-  private static final String COLLECTION_NAME_DOC = "Name of the solr collection to write to.";
 
   public final List<String> zookeeperHosts;
   public final String zookeeperChroot;
-  public final String collectionName;
 
   protected CloudSolrSinkConnectorConfig(Map<String, String> props) {
     super(config(), props);
     this.zookeeperHosts = this.getList(ZOOKEEPER_HOSTS_CONFIG);
     this.zookeeperChroot = this.getString(ZOOKEEPER_CHROOT_CONFIG);
-    this.collectionName = this.getString(COLLECTION_NAME_CONFIG);
   }
 
 
   public static ConfigDef config() {
     return SolrSinkConnectorConfig.config()
-        .define(ZOOKEEPER_HOSTS_CONFIG, ConfigDef.Type.LIST, ConfigDef.Importance.HIGH, ZOOKEEPER_HOSTS_DOC)
-        .define(ZOOKEEPER_CHROOT_CONFIG, ConfigDef.Type.STRING, null, ConfigDef.Importance.HIGH, ZOOKEEPER_CHROOT_DOC)
-        .define(COLLECTION_NAME_CONFIG, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, COLLECTION_NAME_DOC);
+        .define(
+            ConfigKeyBuilder.of(ZOOKEEPER_HOSTS_CONFIG, ConfigDef.Type.LIST)
+            .importance(ConfigDef.Importance.HIGH)
+            .documentation(ZOOKEEPER_HOSTS_DOC)
+            .group(CONNECTION_GROUP)
+            .build()
+        ).define(
+            ConfigKeyBuilder.of(ZOOKEEPER_CHROOT_CONFIG, ConfigDef.Type.STRING)
+                .importance(ConfigDef.Importance.HIGH)
+                .documentation(ZOOKEEPER_CHROOT_DOC)
+                .group(CONNECTION_GROUP)
+                .defaultValue(null)
+                .build()
+        );
   }
 }

src/main/java/com/github/jcustenborder/kafka/connect/solr/HttpSolrSinkConnector.java

@@ -16,10 +16,16 @@
 package com.github.jcustenborder.kafka.connect.solr;
 
 import com.github.jcustenborder.kafka.connect.utils.config.Description;
+import com.github.jcustenborder.kafka.connect.utils.config.DocumentationTip;
+import com.github.jcustenborder.kafka.connect.utils.config.Title;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.connect.connector.Task;
 
+@Title("Standard Solr")
 @Description("This connector is used to connect to write directly to a Solr core.")
+@DocumentationTip("The target collection for this connector is selected by the topic name. " +
+    "`Transformations <https://kafka.apache.org/documentation/#connect_transforms>`_ like the " +
+    "RegexRouter transformation can be used to change the topic name before it is sent to Solr.")
 public class HttpSolrSinkConnector extends SolrSinkConnector {
   @Override
   public Class<? extends Task> taskClass() {
@@ -30,6 +36,4 @@ public Class<? extends Task> taskClass() {
   public ConfigDef config() {
     return HttpSolrSinkConnectorConfig.config();
   }
-
-
 }

src/main/java/com/github/jcustenborder/kafka/connect/solr/HttpSolrSinkConnectorConfig.java

@@ -15,6 +15,7 @@
  */
 package com.github.jcustenborder.kafka.connect.solr;
 
+import com.github.jcustenborder.kafka.connect.utils.config.ConfigKeyBuilder;
 import org.apache.kafka.common.config.ConfigDef;
 
 import java.util.Map;
@@ -44,8 +45,26 @@ public HttpSolrSinkConnectorConfig(Map<String, String> props) {
 
   public static ConfigDef config() {
     return SolrSinkConnectorConfig.config()
-        .define(SOLR_URL_CONFIG, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, SOLR_URL_DOC)
-        .define(SOLR_QUEUE_SIZE_CONFIG, ConfigDef.Type.INT, 100, ConfigDef.Range.between(1, Integer.MAX_VALUE), ConfigDef.Importance.MEDIUM, SOLR_QUEUE_SIZE_DOC)
-        .define(SOLR_THREAD_COUNT_CONFIG, ConfigDef.Type.INT, 1, ConfigDef.Range.between(1, 100), ConfigDef.Importance.MEDIUM, SOLR_THREAD_COUNT_DOC);
+        .define(
+            ConfigKeyBuilder.of(SOLR_URL_CONFIG, ConfigDef.Type.STRING)
+                .importance(ConfigDef.Importance.HIGH)
+                .documentation(SOLR_URL_DOC)
+                .group(SolrSinkConnectorConfig.CONNECTION_GROUP)
+                .build()
+        ).define(
+            ConfigKeyBuilder.of(SOLR_QUEUE_SIZE_CONFIG, ConfigDef.Type.INT)
+                .importance(ConfigDef.Importance.HIGH)
+                .defaultValue(100)
+                .documentation(SOLR_QUEUE_SIZE_DOC)
+                .group(SolrSinkConnectorConfig.INDEXING_GROUP)
+                .build()
+        ).define(
+            ConfigKeyBuilder.of(SOLR_THREAD_COUNT_CONFIG, ConfigDef.Type.INT)
+                .importance(ConfigDef.Importance.MEDIUM)
+                .defaultValue(1)
+                .documentation(SOLR_THREAD_COUNT_DOC)
+                .group(SolrSinkConnectorConfig.INDEXING_GROUP)
+                .build()
+        );
   }
 }

src/main/java/com/github/jcustenborder/kafka/connect/solr/Operations.java

@@ -44,6 +44,12 @@ UpdateRequest addOperation(boolean delete) {
     if (this.config.commitWithin > 0) {
       result.setCommitWithin(this.config.commitWithin);
     }
+    if (this.config.useBasicAuthentication) {
+      result.setBasicAuthCredentials(
+          this.config.username,
+          this.config.password
+      );
+    }
     return result;
   }
 

src/main/java/com/github/jcustenborder/kafka/connect/solr/SolrSinkConnectorConfig.java

@@ -15,6 +15,7 @@
  */
 package com.github.jcustenborder.kafka.connect.solr;
 
+import com.github.jcustenborder.kafka.connect.utils.config.ConfigKeyBuilder;
 import com.google.common.base.Strings;
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;
@@ -29,7 +30,9 @@ class SolrSinkConnectorConfig extends AbstractConfig {
 
   static final String SOLR_USERNAME_DOC = "The username to use for basic authentication.";
   static final String SOLR_PASSWORD_DOC = "The password to use for basic authentication.";
-  static final String SOLR_COMMIT_WITHIN_DOC = "Configures Solr UpdaterRequest for a commit within the requested number of milliseconds .";
+  static final String SOLR_COMMIT_WITHIN_DOC = "Configures Solr UpdaterRequest for a commit within " +
+      "the requested number of milliseconds. -1 disables the commit within setting and relies on " +
+      "the standard Solr commit setting.";
   static final String SOLR_DELETE_DOCUMENTS_DOC = "Flag to determine if the connector should delete documents. General " +
       "practice in Kafka is to treat a record that contains a key with a null value as a delete.";
 
@@ -49,11 +52,43 @@ protected SolrSinkConnectorConfig(ConfigDef configDef, Map<String, String> props
     this.deleteDocuments = this.getBoolean(SOLR_DELETE_DOCUMENTS_CONFIG);
   }
 
+  public static final String AUTHENTICATION_GROUP = "Authentication";
+  public static final String INDEXING_GROUP = "Indexing";
+  public static final String CONNECTION_GROUP = "Connection";
+
   public static ConfigDef config() {
     return new ConfigDef()
-        .define(SOLR_COMMIT_WITHIN_CONFIG, ConfigDef.Type.INT, -1, ConfigDef.Importance.LOW, SOLR_COMMIT_WITHIN_DOC)
-        .define(SOLR_USERNAME_CONFIG, ConfigDef.Type.STRING, "", ConfigDef.Importance.HIGH, SOLR_USERNAME_DOC)
-        .define(SOLR_PASSWORD_CONFIG, ConfigDef.Type.PASSWORD, "", ConfigDef.Importance.HIGH, SOLR_PASSWORD_DOC)
-        .define(SOLR_DELETE_DOCUMENTS_CONFIG, ConfigDef.Type.BOOLEAN, true, ConfigDef.Importance.MEDIUM, SOLR_DELETE_DOCUMENTS_DOC);
+        .define(
+            ConfigKeyBuilder.of(SOLR_USERNAME_CONFIG, ConfigDef.Type.STRING)
+                .defaultValue("")
+                .importance(ConfigDef.Importance.HIGH)
+                .documentation(SOLR_USERNAME_DOC)
+                .group(AUTHENTICATION_GROUP)
+                .build()
+        )
+        .define(
+            ConfigKeyBuilder.of(SOLR_PASSWORD_CONFIG, ConfigDef.Type.PASSWORD)
+                .defaultValue("")
+                .importance(ConfigDef.Importance.HIGH)
+                .documentation(SOLR_PASSWORD_DOC)
+                .group(AUTHENTICATION_GROUP)
+                .build()
+        )
+        .define(
+            ConfigKeyBuilder.of(SOLR_COMMIT_WITHIN_CONFIG, ConfigDef.Type.INT)
+                .defaultValue(-1)
+                .importance(ConfigDef.Importance.LOW)
+                .documentation(SOLR_COMMIT_WITHIN_DOC)
+                .group(INDEXING_GROUP)
+                .build()
+        )
+        .define(
+            ConfigKeyBuilder.of(SOLR_DELETE_DOCUMENTS_CONFIG, ConfigDef.Type.BOOLEAN)
+                .defaultValue(true)
+                .importance(ConfigDef.Importance.MEDIUM)
+                .documentation(SOLR_DELETE_DOCUMENTS_DOC)
+                .group(INDEXING_GROUP)
+                .build()
+        );
   }
 }

src/test/resources/com/github/jcustenborder/kafka/connect/solr/CloudSolrSinkConnectorConfig/authentication.json

@@ -0,0 +1,9 @@
+{
+  "name": "Basic Authentication",
+  "description": "This example will connect to a Solr Cloud cluster using basic authentication.",
+  "config": {
+    "solr.zookeeper.hosts": "zookeeper.example.com:2181",
+    "solr.username": "freddy",
+    "solr.password": "password12345"
+  }
+}

src/test/resources/com/github/jcustenborder/kafka/connect/solr/CloudSolrSinkConnectorConfig/example.json

@@ -0,0 +1,7 @@
+{
+  "name": "Standard",
+  "description": "This example will connect to a Solr Cloud cluster without authentication.",
+  "config": {
+    "solr.zookeeper.hosts": "zookeeper.example.com:2181"
+  }
+}

src/test/resources/com/github/jcustenborder/kafka/connect/solr/HttpSolrSinkConnector/authentication.json

@@ -0,0 +1,9 @@
+{
+  "name": "Basic Authentication",
+  "description": "This example will connect to a Solr Cloud cluster using basic authentication.",
+  "config": {
+    "solr.url": "http://solr.example.com:8993/",
+    "solr.username": "freddy",
+    "solr.password": "password12345"
+  }
+}

src/test/resources/com/github/jcustenborder/kafka/connect/solr/HttpSolrSinkConnector/example.json

@@ -0,0 +1,7 @@
+{
+  "name": "Standard",
+  "description": "This example will connect to a Solr Cloud cluster without authentication.",
+  "config": {
+    "solr.url": "http://solr.example.com:8993/"
+  }
+}