1. JWT token support - #14

2. Optional headers support that could be sent to AEP endpoint - Could be use to support raw ingestion
3. Upgrade gradle wrapper from 5.2.1 to 7.0 - jar could be build without installing java separately
4. Default java build to JDK 11, mark JDK 8 build as deprecated
5. Developer guide to use jwt token for authentication

Author: vivetiwa

DEVELOPER_GUIDE.md

@@ -186,8 +186,9 @@ curl -s -X POST \
 
 #### Authentication Enabled
 
-Use the command below to set up an Sink connector to a Authenticated Streaming Connection:
+Use the command below to set up a Sink connector to a Authenticated Streaming Connection:
 
+1. Using access_token
 ```
 curl -s -X POST \
 -H "Content-Type: application/json" \
@@ -211,6 +212,51 @@ curl -s -X POST \
 }' http://localhost:8083/connectors
 ```
 
+2. Using jwt_token
+ - Convert private.key from adobe console to PKCS8 private using command
+```
+openssl pkcs8 -topk8 -inform PEM -outform DER -in private.key -out private-pkcs8.key -nocrypt
+```
+
+ - Create http connector
+```
+curl -s -X POST \
+-H "Content-Type: application/json" \
+--data '{
+  "name": "aep-auth-sink-connector",
+  "config": {
+    "topics": "connect-test",
+    "tasks.max": 1,
+    "aep.flush.interval.seconds": 1,
+    "aep.flush.bytes.kb": 4,
+    "connector.class": "com.adobe.platform.streaming.sink.impl.AEPSinkConnector",
+    "key.converter.schemas.enable": "false",
+    "value.converter.schemas.enable": "false",
+    "aep.endpoint": "https://dcs.adobedc.net/collection/{DATA_INLET_ID}",
+    "aep.connection.auth.enabled": true,
+    "aep.connection.auth.token.type": "jwt_token",
+    "aep.connection.auth.client.id": "<client_id>",
+    "aep.connection.auth.imsOrg": "<organization-id>",
+    "aep.connection.auth.accountKey": "<technical-account-id>",
+    "aep.connection.auth.filePath": "<path-to-private-pkcs8.key>",
+    "aep.connection.auth.endpoint": "<ims-url>",
+    "aep.connection.endpoint.headers": "<optional-header-that-needs-to-be-passed-to-AEP>"
+    "aep.connection.auth.client.secret": "<client_secret>"
+  }
+}' http://localhost:8083/connectors
+```
+Note - `aep.connection.endpoint.headers` format should be comma separated. 
+Example: To send below 2 http header - 
+1. key: x-adobe-flow-id, value: 341fd4f0-cdec-4912-1ab6-fb54aeb41286
+2. key: x-adobe-dataset-id, value: 3096fbfd5978431948af3ba3
+
+Use `aep.connection.endpoint.headers` value -
+```
+{'x-adobe-flow-id':'341fd4f0-cdec-4912-1ab6-fb54aeb41286','x-adobe-dataset-id': '3096fbfd5978431948af3ba3'}
+```
+
+
+
 #### Batching
 
 Use the command below to set up an Sink connector to batch up requests and reduce the number of network calls 

build.gradle

@@ -13,10 +13,11 @@
 buildscript {
   repositories {
     jcenter()
+    mavenCentral()
   }
 
   dependencies {
-    classpath 'com.github.jengelman.gradle.plugins:shadow:4.0.3'
+    classpath 'com.github.johnrengelman.shadow:com.github.johnrengelman.shadow.gradle.plugin:6.1.0'
   }
 }
 
@@ -34,6 +35,7 @@ description 'Adobe Experience Platform Streaming Connect'
 
 repositories {
   jcenter()
+  mavenCentral()
 }
 
 subprojects {
@@ -55,8 +57,23 @@ subprojects {
     toolVersion '8.10.1'
   }
 
+  var compileJavaVersion = System.getenv("JAVA_COMPILATION_VERSION") == null ? (JavaVersion.current().getMajorVersion() == "" ? "11" :
+    JavaVersion.current().getMajorVersion()) : System.getenv("JAVA_COMPILATION_VERSION")
+
+  ext {
+    branchName = System.getenv("BRANCHNAME") ?: grgit.branch.current().getName()
+    compileVersion = compileJavaVersion + (JavaVersion.toVersion(compileJavaVersion) <= JavaVersion.VERSION_1_8 ? "-deprecated" : "")
+  }
+
+  java {
+    toolchain {
+      languageVersion.set(JavaLanguageVersion.of(JavaVersion.toVersion(compileJavaVersion).getMajorVersion()))
+    }
+  }
+
   repositories {
     jcenter()
+    mavenCentral()
 
     publishing {
       repositories {
@@ -79,12 +96,6 @@ subprojects {
     }
   }
 
-  compileJava {
-    sourceCompatibility = 1.8
-    targetCompatibility = 1.8
-    options.deprecation = true
-  }
-
   compileTestJava {
     options.deprecation = true
   }

gradle/wrapper/gradle-wrapper.jar

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-5.2.1-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.0-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,21 @@
 #!/usr/bin/env sh
 
+#
+# Copyright 2015 the original author or authors.
+#
+# 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
+#
+#      https://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.
+#
+
 ##############################################################################
 ##
 ##  Gradle start up script for UN*X
@@ -28,7 +44,7 @@ APP_NAME="Gradle"
 APP_BASE_NAME=`basename "$0"`
 
 # Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
-DEFAULT_JVM_OPTS=""
+DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
 
 # Use the maximum available, or set MAX_FD != -1 to use that value.
 MAX_FD="maximum"
@@ -109,8 +125,8 @@ if $darwin; then
     GRADLE_OPTS="$GRADLE_OPTS \"-Xdock:name=$APP_NAME\" \"-Xdock:icon=$APP_HOME/media/gradle.icns\""
 fi
 
-# For Cygwin, switch paths to Windows format before running java
-if $cygwin ; then
+# For Cygwin or MSYS, switch paths to Windows format before running java
+if [ "$cygwin" = "true" -o "$msys" = "true" ] ; then
     APP_HOME=`cygpath --path --mixed "$APP_HOME"`
     CLASSPATH=`cygpath --path --mixed "$CLASSPATH"`
     JAVACMD=`cygpath --unix "$JAVACMD"`

gradlew

@@ -1,3 +1,19 @@
+@rem
+@rem Copyright 2015 the original author or authors.
+@rem
+@rem Licensed under the Apache License, Version 2.0 (the "License");
+@rem you may not use this file except in compliance with the License.
+@rem You may obtain a copy of the License at
+@rem
+@rem      https://www.apache.org/licenses/LICENSE-2.0
+@rem
+@rem Unless required by applicable law or agreed to in writing, software
+@rem distributed under the License is distributed on an "AS IS" BASIS,
+@rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+@rem See the License for the specific language governing permissions and
+@rem limitations under the License.
+@rem
+
 @if "%DEBUG%" == "" @echo off
 @rem ##########################################################################
 @rem
@@ -14,7 +30,7 @@ set APP_BASE_NAME=%~n0
 set APP_HOME=%DIRNAME%
 
 @rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
-set DEFAULT_JVM_OPTS=
+set DEFAULT_JVM_OPTS="-Xmx64m" "-Xms64m"
 
 @rem Find java.exe
 if defined JAVA_HOME goto findJavaFromJavaHome

gradlew.bat

@@ -18,5 +18,8 @@ pluginManagement {
 
 rootProject.name = 'experience-platform-streaming-connect'
 
-include 'streaming-connect-common'
-include 'streaming-connect-sink'
+include ':streaming-connect-common'
+include ':streaming-connect-sink'
+
+project(':streaming-connect-common').projectDir = file('streaming-connect-common')
+project(':streaming-connect-sink').projectDir = file('streaming-connect-sink')

settings.gradle

@@ -13,8 +13,8 @@
 apply from: '../dependencies.gradle'
 
 dependencies {
-  libraries.with {
-    compile collections4, commonslang, guava, httpClient, jackson, jmockit, junitJupiter, jjwt, orgJson, slf4j, jaxb
-    testCompile jmockit, junitJupiter
-  }
+    implementation libraries.collections4, libraries.commonslang, libraries.guava,
+                   libraries.httpClient, libraries.jackson, libraries.jmockit, libraries.junitJupiter,
+                   libraries.jjwt, libraries.orgJson, libraries.slf4j, libraries.jaxb
+    testImplementation libraries.jmockit, libraries.junitJupiter
 }

streaming-connect-common/build.gradle

@@ -61,19 +61,21 @@ private static AuthProvider getIMSAuthProvider(Map<String, String> authPropertie
   }
 
   private static AuthProvider getJWTAuthProvider(Map<String, String> authProperties) {
-    String clientId = authProperties.get(AuthUtils.AUTH_CLIENT_ID);
-    String imsOrgId = authProperties.get(AuthUtils.AUTH_IMS_ORG_ID);
-    String technicalAccountKey = authProperties.get(AuthUtils.AUTH_TECHNICAL_ACCOUNT_ID);
-    String filePath = authProperties.get(AuthUtils.AUTH_PRIVATE_KEY_FILE_PATH);
+    final String clientId = authProperties.get(AuthUtils.AUTH_CLIENT_ID);
+    final String clientSecret = authProperties.get(AuthUtils.AUTH_CLIENT_SECRET);
+    final String imsOrgId = authProperties.get(AuthUtils.AUTH_IMS_ORG_ID);
+    final String technicalAccountKey = authProperties.get(AuthUtils.AUTH_TECHNICAL_ACCOUNT_ID);
+    final String filePath = authProperties.get(AuthUtils.AUTH_PRIVATE_KEY_FILE_PATH);
 
     Preconditions.checkNotNull(clientId, "Invalid client Id");
+    Preconditions.checkNotNull(clientSecret, "Invalid client secret.");
     Preconditions.checkNotNull(imsOrgId, "Invalid IMS Org");
     Preconditions.checkNotNull(technicalAccountKey, "Invalid technical account Id");
 
     String endpoint = authProperties.get(AuthUtils.AUTH_ENDPOINT);
     return StringUtils.isEmpty(endpoint) ?
-      new JWTTokenProvider(clientId, imsOrgId, technicalAccountKey, filePath) :
-      new JWTTokenProvider(endpoint, clientId, imsOrgId, technicalAccountKey, filePath);
+      new JWTTokenProvider(clientId, clientSecret, imsOrgId, technicalAccountKey, filePath) :
+      new JWTTokenProvider(endpoint, clientId, clientSecret, imsOrgId, technicalAccountKey, filePath);
   }
 
   private AuthProviderFactory() {}

streaming-connect-common/src/main/java/com/adobe/platform/streaming/auth/impl/AuthProviderFactory.java

@@ -61,35 +61,37 @@ public class JWTTokenProvider extends AbstractAuthProvider {
   private final String imsOrgId;
   private final String technicalAccountKey;
   private final String clientId;
+  private final String clientSecret;
   private final String keyPath;
   private String jwtToken;
 
-  JWTTokenProvider(String endpoint, String imsOrgId, String clientId, String technicalAccountKey,
+  JWTTokenProvider(String endpoint, String clientId, String clientSecret, String imsOrgId, String technicalAccountKey,
     String keyPath) {
-    this(imsOrgId, clientId, technicalAccountKey, keyPath);
+    this(clientId, clientSecret, imsOrgId, technicalAccountKey, keyPath);
     this.endpoint = endpoint;
   }
 
-  JWTTokenProvider(String imsOrgId, String clientId, String technicalAccountKey, String keyPath) {
+  JWTTokenProvider(String clientId, String clientSecret, String imsOrgId, String technicalAccountKey, String keyPath) {
     this.imsOrgId = imsOrgId;
     this.clientId = clientId;
+    this.clientSecret = clientSecret;
     this.technicalAccountKey = technicalAccountKey;
     this.keyPath = keyPath;
   }
 
   @Override
   protected TokenResponse getTokenResponse() throws AuthException {
     LOG.debug("refreshing expired jwtToken: {}", clientId);
-    StringBuffer params = new StringBuffer()
+    StringBuilder params = new StringBuilder()
       .append("&client_id=").append(clientId)
-      .append("&client_secret=").append(technicalAccountKey)
+      .append("&client_secret=").append(clientSecret)
       .append("&jwt_token=").append(getJWTToken());
 
     try {
       return HttpProducer.newBuilder(endpoint).build().post(
         IMS_ENDPOINT_PATH,
         params.toString().getBytes(),
-        ContentType.MULTIPART_FORM_DATA.getMimeType(),
+        ContentType.APPLICATION_FORM_URLENCODED.getMimeType(),
         getContentHandler()
       );
     } catch (HttpException httpException) {
@@ -151,7 +153,7 @@ private boolean isJWTExpired() {
         StandardCharsets.UTF_8.name()
       );
       JSONObject tokenBodyJson = new JSONObject(tokenBody);
-      Long expiresIn = (Long)tokenBodyJson.get(JWT_EXPIRY_KEY);
+      long expiresIn = ((Number)tokenBodyJson.get(JWT_EXPIRY_KEY)).longValue();
       return System.currentTimeMillis() <= (expiresIn - DEFAULT_JWT_TOKEN_UPDATE_THRESHOLD);
     } catch (UnsupportedEncodingException exception) {
       LOG.error("Exception while parsing JWT token", exception);