Merge pull request #29 from RADAR-CNS/v0.5_release

V0.5 release
RADAR-base · Aug 31, 2017 · ac8631f · ac8631f
2 parents 97144de + e88d15c
commit ac8631f
Show file tree

Hide file tree

Showing 25 changed files with 116 additions and 420 deletions.
diff --git a/README.md b/README.md
@@ -14,7 +14,7 @@ repositories {
 }
 
 dependencies {
-    compile group: 'org.radarcns', name: 'radar-commons', version: '0.4.2'
+    compile group: 'org.radarcns', name: 'radar-commons', version: '0.5'
 }
 ```
 
@@ -26,7 +26,7 @@ repositories {
 }
 
 dependencies {
-    testCompile group: 'org.radarcns', name: 'radar-commons-testing', version: '0.4.2'
+    testCompile group: 'org.radarcns', name: 'radar-commons-testing', version: '0.5'
 }
 ```
 
@@ -51,7 +51,7 @@ configurations.all {
 }
 
 dependencies {
-    compile group: 'org.radarcns', name: 'radar-commons', version: '0.5-SNAPSHOT', changing: true
+    compile group: 'org.radarcns', name: 'radar-commons', version: '0.5.1-SNAPSHOT', changing: true
 }
 ```
 

diff --git a/build.gradle b/build.gradle
@@ -37,7 +37,7 @@ allprojects {
     // Configuration                                                             //
     //---------------------------------------------------------------------------//
 
-    version = '0.4.2'
+    version = '0.5'
     group = 'org.radarcns'
     ext.githubRepoName = 'RADAR-CNS/RADAR-Commons'
 
@@ -217,24 +217,14 @@ dependencies {
     // Direct producer uses KafkaAvroSerializer if initialized
     testImplementation (group: 'io.confluent', name: 'kafka-avro-serializer', version: confluentVersion) {
         exclude group: 'com.101tec'
+        exclude group: 'org.slf4j', module: 'slf4j-log4j12'
     }
     testImplementation group: 'org.radarcns', name: 'radar-schemas-commons', version: radarSchemasVersion
     testImplementation group: 'junit', name: 'junit', version: junitVersion
     testImplementation group: 'org.mockito', name: 'mockito-core', version: mockitoVersion
     testImplementation group: 'org.hamcrest', name: 'hamcrest-all', version: hamcrestVersion
     testImplementation group: 'com.squareup.okhttp3', name: 'mockwebserver', version: okhttpVersion
 
-    // For Topic name validation based on Kafka classes
-    testImplementation (group: 'org.apache.kafka', name: 'kafka_2.11', version: kafkaVersion) {
-        exclude group: 'org.apache.kafka', module: 'kafka-clients'
-        exclude group: 'net.sf.jopt-simple'
-        exclude group: 'com.yammer.metrics'
-        exclude group: 'org.scala-lang.modules'
-        exclude group: 'org.slf4j'
-        exclude group: 'com.101tec'
-        exclude group: 'org.apache.zookeeper'
-    }
-
     codacy group: 'com.github.codacy', name: 'codacy-coverage-reporter', version: codacyVersion
 }
 
@@ -369,6 +359,6 @@ artifactoryPublish {
 }
 
 task wrapper(type: Wrapper) {
-    gradleVersion = '3.4.1'
-    distributionUrl distributionUrl.replace("bin", "all")
+    gradleVersion = '4.1'
+    distributionType 'all'
 }
diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
@@ -1,6 +1,5 @@
-#Fri Mar 17 09:52:45 CET 2017
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-3.4.1-all.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-4.1-all.zip
diff --git a/gradlew b/gradlew
@@ -33,11 +33,11 @@ DEFAULT_JVM_OPTS=""
 # Use the maximum available, or set MAX_FD != -1 to use that value.
 MAX_FD="maximum"
 
-warn ( ) {
+warn () {
     echo "$*"
 }
 
-die ( ) {
+die () {
     echo
     echo "$*"
     echo
@@ -155,7 +155,7 @@ if $cygwin ; then
 fi
 
 # Escape application args
-save ( ) {
+save () {
     for i do printf %s\\n "$i" | sed "s/'/'\\\\''/g;1s/^/'/;\$s/\$/' \\\\/" ; done
     echo " "
 }

diff --git a/...cns/producer/rest/BatchedKafkaSender.java → ...radarcns/producer/BatchedKafkaSender.java b/...cns/producer/rest/BatchedKafkaSender.java → ...radarcns/producer/BatchedKafkaSender.java
@@ -14,12 +14,10 @@
  * limitations under the License.
  */
 
-package org.radarcns.producer.rest;
+package org.radarcns.producer;
 
 import org.radarcns.data.Record;
 import org.radarcns.topic.AvroTopic;
-import org.radarcns.producer.KafkaSender;
-import org.radarcns.producer.KafkaTopicSender;
 
 import java.io.IOException;
 import java.util.ArrayList;

diff --git a/...ns/producer/rest/ThreadedKafkaSender.java → ...adarcns/producer/ThreadedKafkaSender.java b/...ns/producer/rest/ThreadedKafkaSender.java → ...adarcns/producer/ThreadedKafkaSender.java
@@ -14,11 +14,10 @@
  * limitations under the License.
  */
 
-package org.radarcns.producer.rest;
+package org.radarcns.producer;
 
 import org.radarcns.data.Record;
-import org.radarcns.producer.KafkaSender;
-import org.radarcns.producer.KafkaTopicSender;
+import org.radarcns.producer.rest.ConnectionState;
 import org.radarcns.topic.AvroTopic;
 import org.radarcns.util.RollingTimeAverage;
 import org.slf4j.Logger;
@@ -34,9 +33,8 @@
 import java.util.concurrent.TimeUnit;
 
 /**
- * Send Avro Records to a Kafka REST Proxy.
- *
- * This queues messages for a specified amount of time and then sends all messages up to that time.
+ * Send Avro Records to a Kafka REST Proxy. This queues messages for a specified amount of time
+ * and then sends all messages up to that time.
  */
 public class ThreadedKafkaSender<K, V> implements KafkaSender<K, V> {
     private static final Logger logger = LoggerFactory.getLogger(ThreadedKafkaSender.class);

diff --git a/src/main/java/org/radarcns/producer/rest/RestSender.java b/src/main/java/org/radarcns/producer/rest/RestSender.java
@@ -17,17 +17,7 @@
 package org.radarcns.producer.rest;
 
 import com.fasterxml.jackson.core.JsonFactory;
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.util.AbstractMap;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Objects;
-import java.util.concurrent.TimeUnit;
+import okhttp3.Headers;
 import okhttp3.HttpUrl;
 import okhttp3.MediaType;
 import okhttp3.Request;
@@ -37,14 +27,25 @@
 import org.radarcns.data.AvroEncoder;
 import org.radarcns.data.Record;
 import org.radarcns.producer.AuthenticationException;
+import org.radarcns.producer.BatchedKafkaSender;
 import org.radarcns.producer.KafkaSender;
 import org.radarcns.producer.KafkaTopicSender;
-import org.radarcns.producer.SchemaRetriever;
+import org.radarcns.producer.ThreadedKafkaSender;
 import org.radarcns.producer.rest.ConnectionState.State;
 import org.radarcns.topic.AvroTopic;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
 /**
  * RestSender sends records to the Kafka REST Proxy. It does so using an Avro JSON encoding. A new
  * sender must be constructed with {@link #sender(AvroTopic)} per AvroTopic. This implementation is
@@ -58,6 +59,8 @@
 @SuppressWarnings("PMD.GodClass")
 public class RestSender<K, V> implements KafkaSender<K, V> {
     private static final Logger logger = LoggerFactory.getLogger(RestSender.class);
+    private static final int LOG_CONTENT_LENGTH = 1024;
+
     public static final String KAFKA_REST_ACCEPT_ENCODING =
             "application/vnd.kafka.v2+json, application/vnd.kafka+json, application/json";
     public static final String KAFKA_REST_ACCEPT_LEGACY_ENCODING =
@@ -73,14 +76,14 @@ public class RestSender<K, V> implements KafkaSender<K, V> {
 
     private HttpUrl schemalessKeyUrl;
     private HttpUrl schemalessValueUrl;
-    private Request isConnectedRequest;
+    private Request.Builder isConnectedRequest;
     private SchemaRetriever schemaRetriever;
     private RestClient httpClient;
     private String acceptType;
     private MediaType contentType;
     private boolean useCompression;
     private final ConnectionState state;
-    private List<Entry<String, String>> additionalHeaders;
+    private Headers additionalHeaders;
 
     /**
      * Construct a RestSender.
@@ -90,11 +93,11 @@ public class RestSender<K, V> implements KafkaSender<K, V> {
      * @param valueEncoder non-null Avro encoder for values
      * @param useCompression use compression to send data
      * @param sharedState shared connection state
-     * @param additionalHeaders
+     * @param additionalHeaders headers to add to requests
      */
     private RestSender(RestClient httpClient, SchemaRetriever schemaRetriever,
             AvroEncoder keyEncoder, AvroEncoder valueEncoder, boolean useCompression,
-            ConnectionState sharedState, List<Entry<String, String>> additionalHeaders) {
+            ConnectionState sharedState, Headers additionalHeaders) {
         this.schemaRetriever = schemaRetriever;
         this.keyEncoder = keyEncoder;
         this.valueEncoder = valueEncoder;
@@ -132,7 +135,7 @@ private void setRestClient(RestClient newClient) {
         try {
             schemalessKeyUrl = HttpUrl.get(newClient.getRelativeUrl("topics/schemaless-key"));
             schemalessValueUrl = HttpUrl.get(newClient.getRelativeUrl("topics/schemaless-value"));
-            isConnectedRequest = newClient.requestBuilder("").head().build();
+            isConnectedRequest = newClient.requestBuilder("").head();
         } catch (MalformedURLException ex) {
             throw new IllegalArgumentException("Schemaless topics do not have a valid URL", ex);
         }
@@ -160,7 +163,7 @@ private synchronized HttpUrl getSchemalessKeyUrl() {
     }
 
     private synchronized Request getIsConnectedRequest() {
-        return isConnectedRequest;
+        return isConnectedRequest.headers(additionalHeaders).build();
     }
 
     public synchronized void setCompression(boolean useCompression) {
@@ -171,11 +174,11 @@ private synchronized boolean hasCompression() {
         return this.useCompression;
     }
 
-    public synchronized List<Entry<String, String>> getHeaders() {
+    public synchronized Headers getHeaders() {
         return additionalHeaders;
     }
 
-    public synchronized void setHeaders(List<Entry<String, String>> additionalHeaders) {
+    public synchronized void setHeaders(Headers additionalHeaders) {
         this.additionalHeaders = additionalHeaders;
     }
 
@@ -243,7 +246,7 @@ public void send(List<Record<L, W>> records) throws IOException {
                     String content = response.body().string();
                     String requestContent = ((TopicRequestBody)request.body()).content();
                     requestContent = requestContent.substring(0,
-                            Math.min(requestContent.length(), 255));
+                            Math.min(requestContent.length(), LOG_CONTENT_LENGTH));
                     logger.error("FAILED to transmit message: {} -> {}...",
                             content, requestContent);
                     throw new IOException("Failed to submit (HTTP status code " + response.code()
@@ -253,7 +256,7 @@ public void send(List<Record<L, W>> records) throws IOException {
                 state.didDisconnect();
                 String requestContent = ((TopicRequestBody)request.body()).content();
                 requestContent = requestContent.substring(0,
-                        Math.min(requestContent.length(), 255));
+                        Math.min(requestContent.length(), LOG_CONTENT_LENGTH));
                 logger.error("FAILED to transmit message:\n{}...", requestContent);
                 throw ex;
             } finally {
@@ -270,7 +273,7 @@ private Request buildRequest(List<Record<L, W>> records) throws IOException {
 
             MediaType currentContentType;
             String currentAcceptType;
-            List<Entry<String, String>> currentHeaders;
+            Headers currentHeaders;
 
             synchronized (RestSender.this) {
                 currentContentType = contentType;
@@ -281,12 +284,9 @@ private Request buildRequest(List<Record<L, W>> records) throws IOException {
             TopicRequestBody requestBody;
             Request.Builder requestBuilder = new Request.Builder()
                     .url(sendToUrl)
+                    .headers(currentHeaders)
                     .addHeader("Accept", currentAcceptType);
 
-            for (Map.Entry<String, String> header : currentHeaders) {
-                requestBuilder.addHeader(header.getKey(), header.getValue());
-            }
-
             if (hasCompression()) {
                 requestBody = new GzipTopicRequestBody(requestData, currentContentType);
                 requestBuilder.addHeader("Content-Encoding", "gzip");
@@ -413,7 +413,7 @@ public static class Builder<K, V> {
         private long timeout = 10;
         private ConnectionState state;
         private ManagedConnectionPool pool;
-        private List<Entry<String, String>> additionalHeaders;
+        private Headers.Builder additionalHeaders = new Headers.Builder();
 
         public Builder<K, V> server(ServerConfig kafkaConfig) {
             this.kafkaConfig = kafkaConfig;
@@ -452,15 +452,15 @@ public Builder<K, V> connectionPool(ManagedConnectionPool pool) {
         }
 
         public Builder<K, V> headers(List<Map.Entry<String, String>> headers) {
-            this.additionalHeaders = headers;
+            additionalHeaders = new Headers.Builder();
+            for (Entry<String, String> header : headers) {
+                additionalHeaders.add(header.getKey(), header.getValue());
+            }
             return this;
         }
 
         public Builder<K, V> addHeader(String header, String value) {
-            if (additionalHeaders == null) {
-                additionalHeaders = new ArrayList<>();
-            }
-            additionalHeaders.add(new AbstractMap.SimpleImmutableEntry<>(header, value));
+            additionalHeaders.add(header, value);
             return this;
         }
 
@@ -474,7 +474,6 @@ public RestSender<K, V> build() {
             }
             ConnectionState useState;
             ManagedConnectionPool usePool;
-            List<Entry<String, String>> useHeaders;
 
             if (state != null) {
                 useState = state;
@@ -486,14 +485,10 @@ public RestSender<K, V> build() {
             } else {
                 usePool = ManagedConnectionPool.GLOBAL_POOL;
             }
-            if (additionalHeaders != null) {
-                useHeaders = additionalHeaders;
-            } else {
-                useHeaders = Collections.emptyList();
-            }
 
             return new RestSender<>(new RestClient(kafkaConfig, timeout, usePool),
-                    retriever, keyEncoder, valueEncoder, compression, useState, useHeaders);
+                    retriever, keyEncoder, valueEncoder, compression, useState,
+                    additionalHeaders.build());
         }
     }
 }
diff --git a/...rg/radarcns/producer/SchemaRetriever.java → ...darcns/producer/rest/SchemaRetriever.java b/...rg/radarcns/producer/SchemaRetriever.java → ...darcns/producer/rest/SchemaRetriever.java
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.radarcns.producer;
+package org.radarcns.producer.rest;
 
 import com.fasterxml.jackson.core.JsonEncoding;
 import com.fasterxml.jackson.core.JsonFactory;
@@ -39,9 +39,6 @@
 import org.apache.avro.Schema.Type;
 import org.apache.avro.generic.GenericContainer;
 import org.radarcns.config.ServerConfig;
-import org.radarcns.producer.rest.ManagedConnectionPool;
-import org.radarcns.producer.rest.ParsedSchemaMetadata;
-import org.radarcns.producer.rest.RestClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;