enable docker support

Author: mguowayfair

Dockerfile

@@ -0,0 +1,16 @@
+FROM adoptopenjdk/openjdk8
+
+ARG VERSION
+
+EXPOSE 32311
+EXPOSE 9990
+EXPOSE 5005
+
+RUN apt-get update && apt-get autoclean && apt-get autoremove && apt-get install vim jq -y
+ENV BROOKLIN_TAR_FILE brooklin-${VERSION}.tgz
+
+COPY ${BROOKLIN_TAR_FILE} /opt/app/
+RUN tar -zxf /opt/app/${BROOKLIN_TAR_FILE} --directory=/opt/app && mv /opt/app/brooklin-${VERSION} /opt/app/brooklin
+WORKDIR /opt/app/brooklin
+
+ENTRYPOINT ["/opt/app/brooklin/bin/brooklin-server-start.sh", "/opt/app/brooklin/config/server.properties"]

build.gradle

@@ -3,10 +3,15 @@ apply from: file('gradle/environment.gradle')
 apply from: file("gradle/dependency-versions.gradle")
 apply from: file("gradle/install-git-hooks.gradle")
 
+apply plugin: 'com.palantir.docker'
+
 buildscript {
   repositories {
     mavenCentral()
   }
+  dependencies {
+    classpath "com.palantir.docker:com.palantir.docker.gradle.plugin:0.26.0"
+  }
   apply from: file('gradle/buildscript.gradle'), to: buildscript
 }
 
@@ -18,7 +23,6 @@ allprojects {
   apply plugin: 'project-report'
   apply plugin: 'checkstyle'
   apply plugin: 'findbugs'
-
   repositories {
     mavenCentral()
     jcenter()
@@ -44,6 +48,14 @@ idea {
   }
 }
 
+docker {
+  name 'brooklin'
+  tags 'latest' // deprecated, use 'tag'
+  dockerfile file('Dockerfile')
+  files file("${project.rootDir}/datastream-tools/build/distributions/${rootProject.name}-${rootProject.version}.tgz")
+  buildArgs([VERSION: rootProject.version])
+}
+
 subprojects {
   apply plugin: 'java'
   apply plugin: 'pegasus'
@@ -396,39 +408,6 @@ project(':datastream-tools') {
     duplicatesStrategy 'exclude'
   }
 
-  tasks.create(name: "copyLibs", group: "distribution", dependsOn: configurations.archives.artifacts, type: Copy) {
-    description = "Copy libs to debug folder"
-
-    // debugLibDir is supposed to provide as a cli parameter
-    // e.g. ./gradlew build -x test -x check -x licenseMain copyLibs
-
-    into "${debugLibDir}/libs/"
-    from configurations.archives.artifacts.files,
-      project(':datastream-bigquery').jar,
-      project(':datastream-bigquery').configurations.runtime,
-      project(':datastream-cloud-storage').jar,
-      project(':datastream-cloud-storage').configurations.runtime,
-      project(':datastream-tools').jar,
-      project(':datastream-tools').configurations.runtime,
-      project(':datastream-server').jar,
-      project(':datastream-server').configurations.runtime,
-      project(':datastream-file-connector').jar,
-      project(':datastream-file-connector').configurations.runtime,
-      project(':datastream-jdbc-connector').jar,
-      project(':datastream-jdbc-connector').configurations.runtime,
-      project(':datastream-directory').jar,
-      project(':datastream-directory').configurations.runtime,
-      project(':datastream-testcommon').jar,
-      project(':datastream-testcommon').configurations.runtime,
-      project(':datastream-common').jar,
-      project(':datastream-common').configurations.runtime,
-      project(':datastream-kafka').jar,
-      project(':datastream-kafka').configurations.runtime,
-      project(':datastream-kafka-connector').jar,
-      project(':datastream-kafka-connector').configurations.runtime
-
-    duplicatesStrategy 'exclude'
-  }
   tasks.create(name: "copyDependentLibs", type: Copy) {
     from (configurations.runtime) {
     }

datastream-jdbc-connector/src/main/java/com/linkedin/datastream/connectors/jdbc/JDBCConnector.java

@@ -95,13 +95,12 @@ public JDBCConnector(VerifiableProperties config) {
         _checkpointStoreTopic = config.getProperty(CONFIG_CHECKPOINT_STORE_TOPIC);
 
         _jdbcUser = config.getString(CONFIG_JDBC_USER);
-        _jdbcUserPassword = "P@ssw0rd";
-//        try {
-//            _jdbcUserPassword = Passwords.get(config.getString(CONFIG_JDBC_CREDENTIAL_NAME));
-//        } catch (IOException e) {
-//            _logger.error("Unable to decrypt password.");
-//            throw new RuntimeException(e);
-//        }
+        try {
+            _jdbcUserPassword = Passwords.get(config.getString(CONFIG_JDBC_CREDENTIAL_NAME));
+        } catch (IOException e) {
+            _logger.error("Unable to decrypt password.");
+            throw new RuntimeException(e);
+        }
 
         _cpMinIdle = config.getInt(CONFIG_CP_MIN_IDLE, 1);
         _cpMaxIdle = config.getInt(CONFIG_CP_MAX_IDLE, 5);

datastream-jdbc-connector/src/main/java/com/linkedin/datastream/connectors/jdbc/JDBCConnectorTask.java

@@ -20,6 +20,7 @@
 import javax.sql.DataSource;
 
 import org.apache.avro.generic.GenericRecord;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -142,9 +143,17 @@ private Long getInitialCheckpoint() {
 
     private String generateStatement() {
         String suffix = " WHERE " + _incrementingColumnName + " > ? ORDER BY " + _incrementingColumnName + " ASC";
-        return (_query != null) ?
-                _query + suffix :
-                "SELECT * FROM " + _table + " WITH (NOLOCK)" + suffix;
+        if (StringUtils.isNotBlank(_query)) {
+            // Assume the query is complete if it has `where`
+            if (_query.toUpperCase().contains("WHERE")) {
+                return _query;
+            }
+            return _query + suffix;
+        }
+        if (StringUtils.isBlank(_table)) {
+            throw new IllegalArgumentException("table name cannot be empty!");
+        }
+        return "SELECT * FROM " + _table + " WITH (NOLOCK)" + suffix;
     }
 
     private void poll() {

datastream-server/src/main/java/com/linkedin/datastream/server/providers/KafkaCustomCheckpointProvider.java

@@ -93,11 +93,6 @@ public KafkaCustomCheckpointProvider(String taskId,
             topicConfig.put("delete.retention.ms", "3600000");
             NewTopic newTopic = new NewTopic(topic, 1, (short) 1).configs(topicConfig);
 
-            Random rand = new Random();
-            int test = rand.nextInt(3);
-            if (test == 1) {
-                throw new InterruptedException("purposely throw an exception for testing");
-            }
             CreateTopicsResult result = adminClient.createTopics(Collections.singletonList(newTopic));
             result.values().get(topic).get();
         } catch (InterruptedException e) {

docker-build.sh

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+
+docker build --build-arg VERSION=1.0.2-21-SNAPSHOT -t brooklin . 

docker-compose.yaml

@@ -0,0 +1,15 @@
+version: '3'
+services:
+  brooklin:
+    image: brooklin
+    environment:
+      JVM_OPTS: "-Xms1024M -Xmx2048M"
+      LOG4J_OPTS: "-Dlog4j.configuration=file:/opt/app/brooklin/config/log4j.properties"
+      JMX_PORT: "9990"
+      DEBUG_OPTS: "-agentlib:jdwp=transport=dt_socket,server=y,suspend=${DEBUG_SUSPEND:-n},address=5005"
+    volumes:
+      - /Users/mg699v/Work/gary-mguo/brooklin-utils/scripts/config:/opt/app/brooklin/config
+    ports:
+      - "32311:32311"
+      - "5005:5005"
+      - "9990:9990"

scripts/brooklin-server-start.sh

@@ -16,8 +16,7 @@ if [ "x$HEAP_OPTS" = "x" ]; then
 fi
 
 # Add the optional debug params, useful for development
-DEBUG_PARAMS=${DEBUG_PARAMS:-}
-EXTRA_ARGS="-name brooklin -loggc ${DEBUG_PARAMS} "
+EXTRA_ARGS="-name brooklin -loggc ${DEBUG_OPTS} "
 
 COMMAND=$1
 case $COMMAND in