update to latest version of commons and http client (#48)

Author: BrandonArp

pom.xml

@@ -75,7 +75,7 @@
     <akka.http.version>2.4.2</akka.http.version>
     <apache.httpclient.version>4.5.1</apache.httpclient.version>
     <apache.httpcore.version>4.4.3</apache.httpcore.version>
-    <arpnetworking.commons.version>1.7.1</arpnetworking.commons.version>
+    <arpnetworking.commons.version>1.9.0</arpnetworking.commons.version>
     <aspectjrt.version>1.8.9</aspectjrt.version>
     <cglib.version>3.2.1</cglib.version>
     <client.protocol.version>0.9.3</client.protocol.version>
@@ -97,8 +97,8 @@
     <logback.steno.version>1.16.0</logback.steno.version>
     <log4j.over.slf4j.version>1.7.12</log4j.over.slf4j.version>
     <metrics.aggregator.protocol.version>1.0.3</metrics.aggregator.protocol.version>
-    <metrics.client.version>0.4.5</metrics.client.version>
-    <metrics.client.http.version>0.1.1</metrics.client.http.version>
+    <metrics.client.version>0.6.0</metrics.client.version>
+    <metrics.client.http.version>0.6.0</metrics.client.http.version>
     <metrics.generator.version>1.1.0</metrics.generator.version>
     <metrics.jvm.extra.version>0.4.2</metrics.jvm.extra.version>
     <mockito.version>1.10.19</mockito.version>
@@ -414,8 +414,8 @@
       <version>${metrics.client.version}</version>
     </dependency>
     <dependency>
-      <groupId>com.arpnetworking.metrics</groupId>
-      <artifactId>sink-http-apache</artifactId>
+      <groupId>com.arpnetworking.metrics.extras</groupId>
+      <artifactId>apache-http-sink-extra</artifactId>
       <version>${metrics.client.http.version}</version>
     </dependency>
     <dependency>

src/main/java/com/arpnetworking/metrics/common/sources/HttpSource.java

@@ -20,9 +20,9 @@
 import akka.actor.Props;
 import akka.actor.UntypedActor;
 import akka.http.javadsl.model.HttpHeader;
+import akka.http.javadsl.model.HttpRequest;
 import akka.http.javadsl.model.HttpResponse;
-import akka.http.scaladsl.model.HttpRequest;
-import akka.http.scaladsl.model.RequestEntity;
+import akka.http.javadsl.model.RequestEntity;
 import akka.japi.Pair;
 import akka.stream.ActorMaterializer;
 import akka.stream.ActorMaterializerSettings;
@@ -104,7 +104,7 @@ public void onReceive(final Object message) throws Exception {
             if (message instanceof RequestReply) {
                 final RequestReply requestReply = (RequestReply) message;
                 // TODO(barp): Fix the ugly HttpRequest cast here due to java vs scala dsl
-                akka.stream.javadsl.Source.single((HttpRequest) requestReply.getRequest())
+                akka.stream.javadsl.Source.single(requestReply.getRequest())
                         .via(_processGraph)
                         .toMat(_sink, Keep.right())
                         .run(_materializer)
@@ -132,7 +132,7 @@ public void onReceive(final Object message) throws Exception {
         /**
          * Constructor.
          *
-         * @param source The {@link CollectdHttpSourceV1} to send notifications through.
+         * @param source The {@link HttpSource} to send notifications through.
          */
         /* package private */ Actor(final HttpSource source) {
             _parser = source._parser;

src/main/java/com/arpnetworking/metrics/mad/Bucket.java

@@ -552,10 +552,8 @@ public Builder setDependentStatistics(final LoadingCache<String, Optional<Immuta
          * @return Steno log compatible representation.
          */
         @LogValue
-        @Override
         public Object toLogValue() {
             return LogValueMapFactory.builder(this)
-                    .put("_super", super.toLogValue())
                     .put("sink", _sink)
                     .put("key", _key)
                     .put("start", _start)

src/main/java/com/arpnetworking/metrics/mad/Main.java

@@ -62,6 +62,7 @@
 import scala.concurrent.duration.Duration;
 
 import java.io.File;
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -219,16 +220,16 @@ private void launchActors(final Injector injector) {
         // Load supplemental routes
         final List<SupplementalRoutes> supplementalHttpRoutes = new ArrayList<>();
         _configuration.getSupplementalHttpRoutesClass().ifPresent(clazz -> {
-                    try {
-                        supplementalHttpRoutes.add(clazz.newInstance());
-                    } catch (final InstantiationException | IllegalAccessException e) {
-                        LOGGER.warn()
-                                .setMessage("Failed to instantiate supplemental http routes")
-                                .addData("supplementalHttpRoutesClass", clazz)
-                                .setThrowable(e)
-                                .log();
-                    }
-                });
+                try {
+                    supplementalHttpRoutes.add(clazz.newInstance());
+                } catch (final InstantiationException | IllegalAccessException e) {
+                    LOGGER.warn()
+                            .setMessage("Failed to instantiate supplemental http routes")
+                            .addData("supplementalHttpRoutesClass", clazz)
+                            .setThrowable(e)
+                            .log();
+                }
+        });
 
         // Create and bind Http server
         final Materializer materializer = ActorMaterializer.create(actorSystem);
@@ -269,7 +270,7 @@ private Injector launchGuice(final ActorSystem actorSystem) {
 
         // Instantiate the metrics factory
         final String sinkHost = "0.0.0.0".equals(_configuration.getHttpHost()) ? "localhost" : _configuration.getHttpHost();
-        final String sinkUrl = "http://" + sinkHost + ":" + _configuration.getHttpPort() + "/metrics/v1/application";
+        final URI sinkUrl = URI.create("http://" + sinkHost + ":" + _configuration.getHttpPort() + "/metrics/v1/application");
         final MetricsFactory metricsFactory = new TsdMetricsFactory.Builder()
                 .setClusterName(_configuration.getMonitoringCluster())
                 .setServiceName("mad")

src/test/java/com/arpnetworking/metrics/mad/AggregatorTest.java

@@ -15,6 +15,8 @@
  */
 package com.arpnetworking.metrics.mad;
 
+import com.arpnetworking.commons.observer.Observable;
+import com.arpnetworking.commons.observer.Observer;
 import com.arpnetworking.metrics.mad.model.DefaultMetric;
 import com.arpnetworking.metrics.mad.model.DefaultRecord;
 import com.arpnetworking.test.TestBeanFactory;
@@ -81,7 +83,7 @@ public void testCloseAfterElapsed() throws InterruptedException {
 
         // Send data to aggregator
         _aggregator.notify(
-                null,
+                OBSERVABLE,
                 new DefaultRecord.Builder()
                         .setMetrics(ImmutableMap.of(
                                 "MyMetric",
@@ -131,7 +133,7 @@ public void testMultipleClusters() throws InterruptedException {
         final DateTime start = DateTime.parse("2015-02-05T00:00:00Z");
 
         _aggregator.notify(
-                null,
+                OBSERVABLE,
                 TestBeanFactory.createRecordBuilder()
                         .setTime(start)
                         .setDimensions(
@@ -148,7 +150,7 @@ public void testMultipleClusters() throws InterruptedException {
                         .build());
 
         _aggregator.notify(
-                null,
+                OBSERVABLE,
                 TestBeanFactory.createRecordBuilder()
                         .setTime(start)
                         .setDimensions(
@@ -211,7 +213,7 @@ public void testMultipleServices() throws InterruptedException {
         final DateTime start = DateTime.parse("2015-02-05T00:00:00Z");
 
         _aggregator.notify(
-                null,
+                OBSERVABLE,
                 TestBeanFactory.createRecordBuilder()
                         .setTime(start)
                         .setDimensions(
@@ -228,7 +230,7 @@ public void testMultipleServices() throws InterruptedException {
                         .build());
 
         _aggregator.notify(
-                null,
+                OBSERVABLE,
                 TestBeanFactory.createRecordBuilder()
                         .setTime(start)
                         .setDimensions(
@@ -291,7 +293,7 @@ public void testMultipleHosts() throws InterruptedException {
         final DateTime start = DateTime.parse("2015-02-05T00:00:00Z");
 
         _aggregator.notify(
-                null,
+                OBSERVABLE,
                 TestBeanFactory.createRecordBuilder()
                         .setTime(start)
                         .setDimensions(
@@ -308,7 +310,7 @@ public void testMultipleHosts() throws InterruptedException {
                         .build());
 
         _aggregator.notify(
-                null,
+                OBSERVABLE,
                 TestBeanFactory.createRecordBuilder()
                         .setTime(start)
                         .setDimensions(
@@ -401,4 +403,11 @@ private List<AggregatedData> getCapturedData(
 
     private static final Quantity ONE = new Quantity.Builder().setValue(1.0).build();
     private static final Quantity TWO = new Quantity.Builder().setValue(2.0).build();
+    private static final Observable OBSERVABLE = new Observable() {
+        @Override
+        public void attach(final Observer observer) { }
+
+        @Override
+        public void detach(final Observer observer) { }
+    };
 }

src/test/java/com/arpnetworking/metrics/mad/sources/MappingSourceTest.java

@@ -94,7 +94,7 @@ public void testMergingObserverInvalidEvent() {
         new MappingSource.MappingObserver(
                 mappingSource,
                 Collections.<Pattern, List<String>>emptyMap())
-                .notify(null, "Not a Record");
+                .notify(OBSERVABLE, "Not a Record");
         Mockito.verifyZeroInteractions(_mockSource);
     }
 
@@ -361,4 +361,14 @@ private static void notify(final Observable observable, final Object event) {
     private Observer _mockObserver;
     private Source _mockSource;
     private MappingSource.Builder _mappingSourceBuilder;
+
+    private static final Observable OBSERVABLE = new Observable() {
+        @Override
+        public void attach(final Observer observer) {
+        }
+
+        @Override
+        public void detach(final Observer observer) {
+        }
+    };
 }