Use LookupSession for async lookups

pom.xml

@@ -47,7 +47,7 @@
         <dependency>
             <groupId>dnsjava</groupId>
             <artifactId>dnsjava</artifactId>
-            <version>3.0.2</version>
+            <version>3.4.2-SNAPSHOT</version>
         </dependency>
         <dependency>
             <groupId>com.google.guava</groupId>

src/main/java/com/spotify/dns/CachingLookupFactory.java

@@ -1,54 +1,54 @@
-/*
- * Copyright (c) 2015 Spotify AB
- *
- * 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.
- */
-
-package com.spotify.dns;
-
-import static java.util.Objects.requireNonNull;
-
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
-import com.google.common.util.concurrent.UncheckedExecutionException;
-import java.util.concurrent.ExecutionException;
-import org.xbill.DNS.Lookup;
-
-/**
- * Caches Lookup instances using a per-thread cache; this is so that different threads will never
- * get the same instance of Lookup. Lookup instances are not thread-safe.
- */
-class CachingLookupFactory implements LookupFactory {
-  private final LookupFactory delegate;
-  private final ThreadLocal<Cache<String, Lookup>> cacheHolder;
-
-  CachingLookupFactory(LookupFactory delegate) {
-    this.delegate = requireNonNull(delegate, "delegate");
-    cacheHolder =
-        ThreadLocal.withInitial(() -> CacheBuilder.newBuilder().build());
-  }
-
-  @Override
-  public Lookup forName(final String fqdn) {
-    try {
-      return cacheHolder.get().get(
-          fqdn,
-          () -> delegate.forName(fqdn)
-      );
-    } catch (ExecutionException e) {
-      throw new DnsException(e);
-    } catch (UncheckedExecutionException e) {
-      throw new DnsException(e);
-    }
-  }
-}
+///*
+// * Copyright (c) 2015 Spotify AB
+// *
+// * 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.
+// */
+//
+//package com.spotify.dns;
+//
+//import static java.util.Objects.requireNonNull;
+//
+//import com.google.common.cache.Cache;
+//import com.google.common.cache.CacheBuilder;
+//import com.google.common.util.concurrent.UncheckedExecutionException;
+//import java.util.concurrent.ExecutionException;
+//import org.xbill.DNS.Lookup;
+//
+///**
+// * Caches Lookup instances using a per-thread cache; this is so that different threads will never
+// * get the same instance of Lookup. Lookup instances are not thread-safe.
+// */
+//class CachingLookupFactory implements LookupFactory {
+//  private final LookupFactory delegate;
+//  private final ThreadLocal<Cache<String, Lookup>> cacheHolder;
+//
+//  CachingLookupFactory(LookupFactory delegate) {
+//    this.delegate = requireNonNull(delegate, "delegate");
+//    cacheHolder =
+//        ThreadLocal.withInitial(() -> CacheBuilder.newBuilder().build());
+//  }
+//
+//  @Override
+//  public Lookup forName(final String fqdn) {
+//    try {
+//      return cacheHolder.get().get(
+//          fqdn,
+//          () -> delegate.forName(fqdn)
+//      );
+//    } catch (ExecutionException e) {
+//      throw new DnsException(e);
+//    } catch (UncheckedExecutionException e) {
+//      throw new DnsException(e);
+//    }
+//  }
+//}

src/main/java/com/spotify/dns/DnsSrvResolver.java

@@ -17,6 +17,7 @@
 package com.spotify.dns;
 
 import java.util.List;
+import java.util.concurrent.CompletionStage;
 
 /**
  * Contract for doing SRV lookups.
@@ -30,5 +31,5 @@ public interface DnsSrvResolver {
    * @return a possibly empty list of matching records
    * @throws DnsException if there was an error doing the DNS lookup
    */
-  List<LookupResult> resolve(String fqdn);
+  CompletionStage<List<LookupResult>> resolve(String fqdn);
 }

src/main/java/com/spotify/dns/DnsSrvResolvers.java

@@ -79,7 +79,7 @@ public DnsSrvResolver build() {
         // or if that's empty, localhost.
         resolver = servers == null ?
                    new ExtendedResolver() :
-                   new ExtendedResolver(servers.toArray(new String[servers.size()]));
+                   new ExtendedResolver(servers.toArray(new String[0]));
       } catch (UnknownHostException e) {
         throw new RuntimeException(e);
       }
@@ -90,9 +90,9 @@ public DnsSrvResolver build() {
 
       LookupFactory lookupFactory = new SimpleLookupFactory(resolver);
 
-      if (cacheLookups) {
-        lookupFactory = new CachingLookupFactory(lookupFactory);
-      }
+//      if (cacheLookups) {
+//        lookupFactory = new CachingLookupFactory(lookupFactory);
+//      }
 
       DnsSrvResolver result = new XBillDnsSrvResolver(lookupFactory);
 

src/main/java/com/spotify/dns/LookupFactory.java

@@ -16,16 +16,16 @@
 
 package com.spotify.dns;
 
-import org.xbill.DNS.Lookup;
+import org.xbill.DNS.lookup.LookupSession;
 
 /**
- * Library-internal interface used for finding or creating {@link Lookup} instances.
+ * Library-internal interface used for finding or creating {@link LookupSession} instances.
  */
 interface LookupFactory {
   /**
-   * Returns a {@link Lookup} instance capable of doing SRV lookups for the supplied FQDN.
+   * Returns a {@link LookupSession} instance capable of doing SRV lookups for the supplied FQDN.
    * @param fqdn the name to do lookups for
    * @return a Lookup instance
    */
-  Lookup forName(String fqdn);
+  LookupSession forName(String fqdn);
 }

src/main/java/com/spotify/dns/MeteredDnsSrvResolver.java

@@ -16,12 +16,14 @@
 
 package com.spotify.dns;
 
+import static com.google.common.base.Throwables.throwIfUnchecked;
 import static java.util.Objects.requireNonNull;
 
 import com.spotify.dns.statistics.DnsReporter;
 import com.spotify.dns.statistics.DnsTimingContext;
 
 import java.util.List;
+import java.util.concurrent.CompletionStage;
 
 /**
  * Tracks metrics for DnsSrvResolver calls.
@@ -36,27 +38,28 @@ class MeteredDnsSrvResolver implements DnsSrvResolver {
   }
 
   @Override
-  public List<LookupResult> resolve(String fqdn) {
+  public CompletionStage<List<LookupResult>> resolve(String fqdn) {
     // Only catch and report RuntimeException to avoid Error's since that would
     // most likely only aggravate any condition that causes them to be thrown.
 
     final DnsTimingContext resolveTimer = reporter.resolveTimer();
 
-    final List<LookupResult> result;
+    return delegate
+        .resolve(fqdn)
+        .handle(
+            (result, error) -> {
+              resolveTimer.stop();
+              if (error == null) {
+                if (result.isEmpty()) {
+                  reporter.reportEmpty();
+                }
 
-    try {
-      result = delegate.resolve(fqdn);
-    } catch (RuntimeException error) {
-      reporter.reportFailure(error);
-      throw error;
-    } finally {
-      resolveTimer.stop();
-    }
-
-    if (result.isEmpty()) {
-      reporter.reportEmpty();
-    }
-
-    return result;
+                return result;
+              } else {
+                reporter.reportFailure(error);
+                throwIfUnchecked(error);
+                throw new RuntimeException(error);
+              }
+            });
   }
 }

src/main/java/com/spotify/dns/RetainingDnsSrvResolver.java

@@ -23,6 +23,7 @@
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import java.util.List;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -46,28 +47,27 @@ class RetainingDnsSrvResolver implements DnsSrvResolver {
   }
 
   @Override
-  public List<LookupResult> resolve(final String fqdn) {
+  public CompletionStage<List<LookupResult>> resolve(final String fqdn) {
     requireNonNull(fqdn, "fqdn");
-
-    try {
-      final List<LookupResult> nodes = delegate.resolve(fqdn);
-
-      // No nodes resolved? Return stale data.
-      if (nodes.isEmpty()) {
-        List<LookupResult> cached = cache.getIfPresent(fqdn);
-        return (cached != null) ? cached : nodes;
-      }
-
-      cache.put(fqdn, nodes);
-
-      return nodes;
-    } catch (Exception e) {
-      if (cache.getIfPresent(fqdn) != null) {
-        return cache.getIfPresent(fqdn);
+    return delegate.resolve(fqdn).handle((nodes, e) -> {
+      if (e == null){
+        // No nodes resolved? Return stale data.
+        if (nodes.isEmpty()) {
+          List<LookupResult> cached = cache.getIfPresent(fqdn);
+          return (cached != null) ? cached : nodes;
+        }
+
+        cache.put(fqdn, nodes);
+
+        return nodes;
+      } else{
+        if (cache.getIfPresent(fqdn) != null) {
+          return cache.getIfPresent(fqdn);
+        }
+
+        throwIfUnchecked(e);
+        throw new RuntimeException(e);
       }
-
-      throwIfUnchecked(e);
-      throw new RuntimeException(e);
-    }
+    });
   }
 }

src/main/java/com/spotify/dns/ServiceResolvingChangeNotifier.java

@@ -19,7 +19,6 @@
 import static java.util.Objects.requireNonNull;
 
 import com.google.common.collect.ImmutableSet;
-import java.util.List;
 import java.util.Set;
 import java.util.function.Function;
 import org.slf4j.Logger;
@@ -53,7 +52,7 @@ class ServiceResolvingChangeNotifier<T> extends AbstractChangeNotifier<T>
    * and put into a set. The set will then be compared to the previous set and if a
    * change is detected, the notifier will fire.
    *
-   * <p>An optional {@link ErrorHandler} can be used to reacto on {@link DnsException}s thrown
+   * <p>An optional {@link ErrorHandler} can be used to react on {@link DnsException}s thrown
    * by the {@link DnsSrvResolver}.
    *
    * @param resolver            The resolver to use.
@@ -88,45 +87,42 @@ public void run() {
       return;
     }
 
-    final List<LookupResult> nodes;
-    try {
-      nodes = resolver.resolve(fqdn);
-    } catch (DnsException e) {
-      if (errorHandler != null) {
-        errorHandler.handle(fqdn, e);
+    resolver.resolve(fqdn).whenComplete((nodes, e) -> {
+      if (e instanceof DnsException) {
+        if (errorHandler != null) {
+          errorHandler.handle(fqdn, (DnsException) e);
+        }
+        log.error(e.getMessage(), e);
+        fireIfFirstError();
+      } else if (e != null) {
+        log.error(e.getMessage(), e);
+        fireIfFirstError();
+      } else {
+        final Set<T> current;
+        try {
+          ImmutableSet.Builder<T> builder = ImmutableSet.builder();
+          for (LookupResult node : nodes) {
+            T transformed = resultTransformer.apply(node);
+            builder.add(requireNonNull(transformed, "transformed"));
+          }
+          current = builder.build();
+        } catch (Exception transformerException) {
+          log.error(transformerException.getMessage(), transformerException);
+          fireIfFirstError();
+          return;
+        }
+
+        if (ChangeNotifiers.isNoLongerInitial(current, records) || !current.equals(records)) {
+          // This means that any subsequent DNS error will be ignored and the existing result will be kept
+          waitingForFirstEvent = false;
+          final ChangeNotification<T> changeNotification =
+                  newChangeNotification(current, records);
+          records = current;
+
+          fireRecordsUpdated(changeNotification);
+        }
       }
-      log.error(e.getMessage(), e);
-      fireIfFirstError();
-      return;
-    } catch (Exception e) {
-      log.error(e.getMessage(), e);
-      fireIfFirstError();
-      return;
-    }
-
-    final Set<T> current;
-    try {
-      ImmutableSet.Builder<T> builder = ImmutableSet.builder();
-      for (LookupResult node : nodes) {
-        T transformed = resultTransformer.apply(node);
-        builder.add(requireNonNull(transformed, "transformed"));
-      }
-      current = builder.build();
-    } catch (Exception e) {
-      log.error(e.getMessage(), e);
-      fireIfFirstError();
-      return;
-    }
-
-    if (ChangeNotifiers.isNoLongerInitial(current, records) || !current.equals(records)) {
-      // This means that any subsequent DNS error will be ignored and the existing result will be kept
-      waitingForFirstEvent = false;
-      final ChangeNotification<T> changeNotification =
-          newChangeNotification(current, records);
-      records = current;
-
-      fireRecordsUpdated(changeNotification);
-    }
+    });
   }
 
   private void fireIfFirstError() {