SDC-1855: Couchbase destination

Change-Id: I003a32471597814898e8b67671cd60b9c00ec472 Reviewed-on: https://review.streamsets.net/15228 Reviewed-by: Jarcec Cecho <jarcec@streamsets.com>
edp963 · Jun 27, 2018 · fbd143b · fbd143b
1 parent 03e493a
commit fbd143b
Show file tree

Hide file tree

Showing 15 changed files with 1,288 additions and 0 deletions.
diff --git a/couchbase-lib/pom.xml b/couchbase-lib/pom.xml
@@ -0,0 +1,104 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Copyright 2018 StreamSets Inc.
+
+  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 the License. See accompanying LICENSE file.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+      <groupId>com.streamsets</groupId>
+      <artifactId>streamsets-datacollector-stage-lib-parent</artifactId>
+      <version>3.4.0-SNAPSHOT</version>
+      <relativePath>../stage-lib-parent</relativePath>
+    </parent>
+    <artifactId>streamsets-datacollector-couchbase-lib</artifactId>
+    <version>3.4.0-SNAPSHOT</version>
+    <description>StreamSets Data Collector Couchbase Library</description>
+    <name>StreamSets Data Collector Couchbase Library</name>
+    <packaging>jar</packaging> 
+    <!-- Core StreamSets Dependencies for Stage Libraries -->
+    <dependencies>
+    <dependency>
+      <groupId>org.codehaus.jettison</groupId>
+      <artifactId>jettison</artifactId>
+      <version>1.3.7</version>
+    </dependency>
+    <dependency>
+      <groupId>com.couchbase.client</groupId>
+      <artifactId>java-client</artifactId>
+      <version>2.5.5</version>
+      <scope>provided</scope>
+    </dependency>
+        <dependency>
+      <groupId>com.streamsets</groupId>
+      <artifactId>streamsets-datacollector-api</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.streamsets</groupId>
+      <artifactId>streamsets-datacollector-commonlib</artifactId>
+      <scope>provided</scope>
+    </dependency>
+     <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
+      <version>${slf4j.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-log4j12</artifactId>
+      <version>${slf4j.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
+     <!-- Test Dependencies -->
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.seleniumhq.selenium</groupId>
+      <artifactId>selenium-java</artifactId>
+      <scope>test</scope>
+      <version>2.44.0</version>
+    </dependency>
+        <dependency>
+      <groupId>com.opera</groupId>
+      <artifactId>operadriver</artifactId>
+      <scope>test</scope>
+      <version>1.5</version>
+      <exclusions>
+        <exclusion>
+          <groupId>org.seleniumhq.selenium</groupId>
+          <artifactId>selenium-remote-driver</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-deploy-plugin</artifactId>
+        <version>2.8.2</version>
+        <configuration>
+          <skip>true</skip>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+</project>
diff --git a/...n/java/com/streamsets/pipeline/stage/destination/couchbase/CouchbaseConnectorDTarget.java b/...n/java/com/streamsets/pipeline/stage/destination/couchbase/CouchbaseConnectorDTarget.java
@@ -0,0 +1,239 @@
+/*
+ * Copyright 2015 StreamSets Inc.
+ *
+ * Licensed under the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 the License.
+ */
+package com.streamsets.pipeline.stage.destination.couchbase;
+
+import com.streamsets.pipeline.api.ConfigDef;
+import com.streamsets.pipeline.api.ConfigGroups;
+import com.streamsets.pipeline.api.GenerateResourceBundle;
+import com.streamsets.pipeline.api.StageDef;
+import com.streamsets.pipeline.api.ValueChooserModel;
+import com.streamsets.pipeline.lib.el.RecordEL;
+import com.streamsets.pipeline.lib.el.TimeEL;
+
+@StageDef(
+    version = 1,
+    label = "Couchbase",
+    description = "Couchbase Destination",
+    icon = "couchbase.png",
+    recordsByRef = true,
+    onlineHelpRefUrl = ""
+)
+
+@ConfigGroups(value = Groups.class)
+@GenerateResourceBundle
+
+public class CouchbaseConnectorDTarget extends CouchbaseConnectorTarget {
+
+    @ConfigDef(
+        required = true,
+        type = ConfigDef.Type.STRING,
+        defaultValue = "localhost:8091",
+        label = "URL",
+        displayPosition = 10,
+        description = "The URL endpoint of the Couchbase NoSQL Database Cluster",
+        group = "COUCHBASE_TARGET"
+    )
+    public String URL;
+
+    /** {@inheritDoc} */
+    @Override
+    public String getURL() {
+      return URL;
+    }
+
+      @ConfigDef(
+        required = true,
+        type = ConfigDef.Type.STRING,
+        defaultValue = "",
+        label = "Bucket",
+        displayPosition = 20,
+        description = "Couchbase Destination Bucket to ingesting data",
+        group = "COUCHBASE_TARGET"
+      )
+    public String bucket;
+
+    /** {@inheritDoc} */
+    @Override
+    public String getBucket() {
+      return bucket;
+    }
+
+    //Couchbase version
+
+    @ConfigDef(
+        required = true,
+        type = ConfigDef.Type.MODEL,
+        defaultValue = "VERSION4",
+        label = "Couchbase NoSQL Database Version",
+        description = "Specify the version of the Couchbase NoSQL Database",
+        displayPosition = 30,
+        group = "COUCHBASE_TARGET"
+    )
+    @ValueChooserModel(CouchbaseVersionChooserValues.class)
+    public CouchbaseVersionTypes version = CouchbaseVersionTypes.VERSION4; //Default
+
+      /** {@inheritDoc} */
+    @Override
+    public CouchbaseVersionTypes getCouchbaseVersion() {
+      return version;
+    }
+
+    //Version 4 Config  
+
+    @ConfigDef(
+        required = true,
+        type = ConfigDef.Type.STRING,
+        defaultValue = "",
+        label = "Password (for Bucket)",
+        displayPosition = 40,
+        description = "Password of the Couchbase Version 4 Bucket",
+        dependsOn = "version",
+        triggeredByValue = "VERSION4",
+        group = "COUCHBASE_TARGET"
+    )
+    public String bucketPassword;
+
+    /** {@inheritDoc} */
+    @Override
+    public String getBucketPassword() {
+      return bucketPassword;
+    }
+
+    //Version 5 Config
+
+    @ConfigDef(
+        required = true,
+        type = ConfigDef.Type.STRING,
+        defaultValue = "",
+        label = "Couchbase User Name",
+        displayPosition = 40,
+        description = "Specify a Couchbase user name for connecting the bucket",
+        dependsOn = "version",
+        triggeredByValue = "VERSION5",
+        group = "COUCHBASE_TARGET"
+    )
+    public String userName;
+
+    /** {@inheritDoc} */
+    @Override
+    public String getUserName() {
+      return userName;
+    }
+
+    @ConfigDef(
+        required = true,
+        type = ConfigDef.Type.STRING,
+        defaultValue = "",
+        label = "Couchbase User Password",
+        displayPosition = 50,
+        description = "Specify a password for the Couchbase user name",
+        dependsOn = "version",
+        triggeredByValue = "VERSION5",
+        group = "COUCHBASE_TARGET"
+    )
+    public String userPassword;
+
+    /** {@inheritDoc} */
+    @Override
+    public String getUserPassword() {
+      return userPassword;
+    }
+
+    //Document Key Selection
+
+    @ConfigDef(
+        required = true,
+        type = ConfigDef.Type.MODEL,
+        defaultValue = "FIELD",
+        label = "Document Key Type",
+        description = "Specify the type of document key",
+        displayPosition = 50,
+        group = "COUCHBASE_TARGET"
+    )
+    @ValueChooserModel(CouchbaseDocumentKeyChooserValues.class)
+    public CouchbaseDocumentKeyTypes documentType = CouchbaseDocumentKeyTypes.FIELD; //Default
+
+      /** {@inheritDoc} */
+    @Override
+    public CouchbaseDocumentKeyTypes getDocumentType() {
+      return documentType;
+    }
+
+    @ConfigDef(
+      required = true,
+      type = ConfigDef.Type.STRING,
+      defaultValue = "",
+      label = "Unique Document Key Field",
+      displayPosition = 60,
+      description = "A field in the document/data which will be used as the unique document key in Couchbase",
+      group = "COUCHBASE_TARGET",
+      dependsOn = "documentType",
+      triggeredByValue = "FIELD"
+    )
+
+    public String documentKey;
+
+    /** {@inheritDoc} */
+    @Override
+    public String getDocumentKey() {
+      return documentKey;
+    }
+
+    @ConfigDef(
+      required = true,
+      type = ConfigDef.Type.STRING,
+      defaultValue = "",
+      label = "Custom Document Key",
+      displayPosition = 60,
+      description = "Custom Document Key. Use the Expression Language to define a custom document key. "
+              + "Example: str:concat('customer::', records:value('/customerID'))",
+      group = "COUCHBASE_TARGET",
+      dependsOn = "documentType",
+      triggeredByValue = "CUSTOM_DOC_KEY",
+      elDefs = {RecordEL.class, TimeEL.class},
+      evaluation = ConfigDef.Evaluation.EXPLICIT
+    )
+
+    public String customDocumentKey;
+
+    /** {@inheritDoc} */
+    @Override
+    public String getCustomDocumentKey() {
+      return customDocumentKey;
+    }
+
+    @ConfigDef(
+      required = false,
+      type = ConfigDef.Type.BOOLEAN,
+      defaultValue = "false",
+      label = "Generate unique Document Key",
+      displayPosition = 70,
+      description = "Generate a unique document key if document key field cannot be set",
+      group = "COUCHBASE_TARGET"
+    )
+
+    public boolean generateDocumentKey;
+
+    /** {@inheritDoc} */
+    @Override
+    public boolean generateDocumentKey() {
+        return generateDocumentKey;
+    }
+}