child document support and few minor bug fixes

Author: bkatwal

config/bkSolrSinkConnector.properties

@@ -18,6 +18,11 @@
 name=SimpleKafkaSolrSinkConnector
 topics=<your toipic name>
 solr.collection=<your collection name>
-solr.zkhosts=localhost:2181,localhost:2183,localhost:2182
+#solr.url=localhost:8983/solr
+#solr.mode=STANDALONE
+solr.url=localhost:2182,localhost:2183,localhost:2184
+solr.mode=CLOUD
 connector.class=com.bkatwal.kafkaproject.SolrSinkConnector
 tasks.max=1
+commit.within.ms=10
+

config/bkworker.properties

@@ -1,5 +1,3 @@
-
-
 ##
 # Copyright 2018 BIkas Katwal.
 #
@@ -15,22 +13,17 @@
 # License for the specific language governing permissions and limitations under
 # the License.
 ##
-
-bootstrap.servers=localhost:9092
+bootstrap.servers=localhost:9093
 header.converter=org.apache.kafka.connect.storage.StringConverter
-
 key.converter=org.apache.kafka.connect.storage.StringConverter
 value.converter=org.apache.kafka.connect.json.JsonConverter
-
 internal.key.converter=org.apache.kafka.connect.storage.StringConverter
 internal.value.converter=org.apache.kafka.connect.json.JsonConverter
-
 #to deserialize plain JSON data keep schemas.enable=false
 internal.key.converter.schemas.enable=false
 key.converter.schemas.enable=false
 value.converter.schemas.enable=false
 internal.value.converter.schemas.enable=false
-
 offset.storage.file.filename=/tmp/connect.offsets
 # Flush much faster than normal, which is useful for testing/debugging
 offset.flush.interval.ms=10000

pom.xml

@@ -33,7 +33,13 @@
             <artifactId>solr-solrj</artifactId>
             <version>7.3.0</version>
         </dependency>
-
+        <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>1.18.4</version>
+            <scope>provided</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>connect-api</artifactId>

src/main/java/com/bkatwal/kafkaproject/SolrSinkConnectorConfig.java

@@ -1,80 +1,93 @@
 /**
  * Copyright 2018 Bikas Katwal.
  *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not
- * use this file except in compliance with the License. You may obtain a copy of
- * the License at
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
  *
  * http://www.apache.org/licenses/LICENSE-2.0
  *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations under
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
  * the License.
  **/
 
 package com.bkatwal.kafkaproject;
 
+import com.bkatwal.kafkaproject.utils.SolrMode;
+import java.util.Map;
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigDef.Importance;
 import org.apache.kafka.common.config.ConfigDef.Type;
 
-import java.util.Map;
-
 
 public class SolrSinkConnectorConfig extends AbstractConfig {
 
-    private static final String TOPIC_DOC = "Topic from where data needs to be read.";
+  public static final String COLLECTION_CONFIG = "solr.collection";
+  private static final String COLLECTION_DOC = "Solr Collection name to which data need to be be written";
+
+  public static final String SOLRURL_CONFIG = "solr.url";
+  private static final String SOLRURL_DOC =
+      "Comma separated zookeeper hosts, eg: localhost:2181,localhost:2182,localhost:2183" +
+          " or it could be standalone solr mode too, ex: localhost:8983/solr";
+
+  public static final String USERNAME_CONFIG = "solr.username";
+  private static final String USERNAME_DOC = "username to connect to solr.";
 
-    public static final String COLLECTION_CONFIG = "solr.collection";
-    private static final String COLLECTION_DOC = "Solr Collection name to which data need to be be written";
+  public static final String PASSWORD_CONFIG = "solr.password";
+  private static final String PASSWORD_DOC = "password to connect to solr.";
 
-    public static final String ZKHOSTS_CONFIG = "solr.zkhosts";
-    private static final String ZKHOSTS_DOC = "Comma separated zookeeper hosts, eg: localhost:2181,localhost:2182,localhost:2183";
+  public static final String COMMIT_WITHIN_MS = "commit.within.ms";
+  private static final String COMMIT_WITHIN_MS_DOC = "solr commit within milli seconds param.";
 
-    public static final String USERNAME_CONFIG = "solr.username";
-    private static final String USERNAME_DOC = "username to connect to solr.";
+  public static final String SOLRMODE_CONFIG = "solr.mode";
+  private static final String SOLRMODE_DOC = "solr mode can be STANDALONE/CLOUD";
 
-    public static final String PASSWORD_CONFIG = "solr.password";
-    private static final String PASSWORD_DOC = "password to connect to solr.";
 
+  public SolrSinkConnectorConfig(ConfigDef config, Map<String, String> parsedConfig) {
+    super(config, parsedConfig);
+  }
 
-    public SolrSinkConnectorConfig(ConfigDef config, Map<String, String> parsedConfig) {
-        super(config, parsedConfig);
-    }
+  public SolrSinkConnectorConfig(Map<String, String> parsedConfig) {
+    this(conf(), parsedConfig);
+  }
 
-    public SolrSinkConnectorConfig(Map<String, String> parsedConfig) {
-        this(conf(), parsedConfig);
-    }
+  public static ConfigDef conf() {
+    return new ConfigDef()
+        .define(COLLECTION_CONFIG, Type.STRING, Importance.HIGH, COLLECTION_DOC)
+        .define(SOLRURL_CONFIG, Type.STRING, Importance.HIGH, SOLRURL_DOC)
+        .define(USERNAME_CONFIG, Type.STRING, "", Importance.MEDIUM, USERNAME_DOC)
+        .define(PASSWORD_CONFIG, Type.PASSWORD, "", Importance.MEDIUM, PASSWORD_DOC)
+        .define(SOLRMODE_CONFIG, Type.STRING, "", Importance.MEDIUM, SOLRMODE_DOC)
+        .define(COMMIT_WITHIN_MS, Type.INT, 10, Importance.MEDIUM, COMMIT_WITHIN_MS_DOC);
+  }
 
-    public static ConfigDef conf() {
-        return new ConfigDef()
-                .define(COLLECTION_CONFIG, Type.STRING, Importance.HIGH, COLLECTION_DOC)
-                .define(ZKHOSTS_CONFIG, Type.STRING, Importance.HIGH, ZKHOSTS_DOC)
-                .define(USERNAME_CONFIG, Type.STRING, "", Importance.MEDIUM, USERNAME_DOC)
-                .define(PASSWORD_CONFIG, Type.PASSWORD, "", Importance.MEDIUM, PASSWORD_DOC);
-    }
 
+  public String getCollectionConfig() {
+    return this.getString(COLLECTION_CONFIG);
+  }
 
-    public  String getCollectionConfig() {
-        return this.getString(COLLECTION_CONFIG);
-    }
 
+  public String getSolrURLConfig() {
+    return this.getString(SOLRURL_CONFIG);
+  }
 
-    public  String getZkhostsConfig() {
-        return this.getString(ZKHOSTS_CONFIG);
-    }
 
+  public String getUsernameConfig() {
+    return this.getString(USERNAME_CONFIG);
+  }
 
-    public  String getUsernameConfig() {
-        return this.getString(USERNAME_CONFIG);
-    }
 
+  public String getPasswordConfig() {
+    return this.getString(PASSWORD_CONFIG);
+  }
 
-    public  String getPasswordConfig() {
-        return this.getString(PASSWORD_CONFIG);
-    }
+  public SolrMode getSolrModeConfig() {
+    return SolrMode.valueOf(this.getString(SOLRMODE_CONFIG));
+  }
 
+  public int getCommitWithinMs(){
+    return getInt(COMMIT_WITHIN_MS);
+  }
 }

src/main/java/com/bkatwal/kafkaproject/SolrSinkTask.java

@@ -1,113 +1,117 @@
 /**
  * Copyright 2018 Bikas Katwal.
  *
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not
- * use this file except in compliance with the License. You may obtain a copy of
- * the License at
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
  *
  * http://www.apache.org/licenses/LICENSE-2.0
  *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations under
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
  * the License.
  **/
 
 package com.bkatwal.kafkaproject;
 
+import com.bkatwal.kafkaproject.utils.PlainJsonSolrDocMappersImpl;
 import com.bkatwal.kafkaproject.utils.SinkService;
+import com.bkatwal.kafkaproject.utils.SolrClientFactory;
 import com.bkatwal.kafkaproject.utils.SolrSinkService;
+import java.util.Collection;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.apache.kafka.connect.sink.SinkTask;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Collection;
-import java.util.Map;
 
+@Slf4j
 public class SolrSinkTask extends SinkTask {
-    private static Logger log = LoggerFactory.getLogger(SolrSinkTask.class);
 
-    private SinkService sinkService;
-    private SolrSinkConnectorConfig config;
+  private SinkService<String, SinkRecord> sinkService;
+  private SolrSinkConnectorConfig config;
 
-    @Override
-    public String version() {
-        return VersionUtil.getVersion(SolrSinkTask.class);
-    }
+  @Override
+  public String version() {
+    return VersionUtil.getVersion(SolrSinkTask.class);
+  }
 
-    @Override
-    public void start(Map<String, String> configMap) {
-        config = new SolrSinkConnectorConfig(configMap);
-        sinkService = new SolrSinkService(config);
-        log.debug("Created config:" + config);
+  @Override
+  public void start(Map<String, String> configMap) {
+    config = new SolrSinkConnectorConfig(configMap);
 
-    }
+    sinkService = SolrSinkService.builder().collection(config.getCollectionConfig())
+        .commitWithinMs(config.getCommitWithinMs())
+        .jsonSolrDocMapper(new PlainJsonSolrDocMappersImpl()).solrClient(SolrClientFactory
+            .getClient(config.getSolrModeConfig(), config.getSolrURLConfig())).build();
 
-    @Override
-    public void put(Collection<SinkRecord> kafkaRecords){
+    log.debug("Created config: {}", config);
 
-        for (SinkRecord record : kafkaRecords) {
-            String id = record.key() != null ? record.key().toString() : null;
+  }
 
-            Schema valueSchema = record.valueSchema();
+  @Override
+  public void put(Collection<SinkRecord> kafkaRecords) {
 
-            //not a plain json data/schema less data
-            //Expecting schema less record
-            //TODO handle schema based record later
-            if (valueSchema == null) {
+    for (SinkRecord record : kafkaRecords) {
+      String id = record.key() != null ? record.key().toString() : null;
 
-                Map<String, Object> jsonValueMap = (Map<String, Object>) record.value();
+      Schema valueSchema = record.valueSchema();
 
-                Object delVal = jsonValueMap.get("_delete_");
+      //not a plain json data/schema less data
+      //Expecting schema less record
+      if (valueSchema != null) {
+        log.error(
+            "Check if record in topic is plain json data and value is schema less. Set schema.enable=false for value.");
+        throw new ConnectException(
+            "Check if record in topic is plain json data and value is schema less. Set schema.enable=false for value.");
+      }
 
-                //delete the field "_delete_" after reading the value from it
-                jsonValueMap.remove("_delete_");
+      Map<String, Object> jsonValueMap = (Map<String, Object>) record.value();
 
-                //if _delete_ field is passed as false in value or if value is null, respective
-                // doc will be deleted from solr
-                if (isDeleteRequest(delVal) || record.value() == null) {
+      Object delVal = jsonValueMap.get("_delete_");
 
-                    sinkService.deleteById(id);
-                    
-                } else {
+      //delete the field "_delete_" after reading the value from it
+      jsonValueMap.remove("_delete_");
 
-                    sinkService.insert(id, record);
+      //if _delete_ field is passed as false in value or if value is null, respective
+      // doc will be deleted from solr
+      if (isDeleteRequest(delVal) || record.value() == null) {
 
-                }
-            }
-            log.error("Check if record in topic is plain json data and value is schema less. Set schema.enable=false for value.");
+        sinkService.deleteById(id);
 
-        }
+      } else {
 
-    }
+        sinkService.insert(id, record);
 
-    private boolean isDeleteRequest(Object _delete_) {
-        return _delete_ == null ? false : _delete_ instanceof String ?
-                Boolean.parseBoolean((String) _delete_) : (Boolean) _delete_;
+      }
     }
 
-    @Override
-    public void flush(Map<TopicPartition, OffsetAndMetadata> map) {
+  }
 
-    }
+  private boolean isDeleteRequest(Object delete) {
+    return delete != null && (delete instanceof String ?
+        Boolean.parseBoolean((String) delete) : (Boolean) delete);
+  }
 
+  @Override
+  public void flush(Map<TopicPartition, OffsetAndMetadata> map) {
+  }
 
-    @Override
-    public void stop() {
-        sinkService.stop();
-    }
 
+  @Override
+  public void stop() {
+    sinkService.stop();
+  }
 
-    public void setConfig(SolrSinkConnectorConfig config) {
-        this.config = config;
-    }
 
-    public SolrSinkConnectorConfig getConfig() {
-        return config;
-    }
+  public void setConfig(SolrSinkConnectorConfig config) {
+    this.config = config;
+  }
+
+  public SolrSinkConnectorConfig getConfig() {
+    return config;
+  }
 }