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</version>
         </dependency>
         <dependency>
             <groupId>com.google.guava</groupId>
@@ -57,7 +57,7 @@
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
-            <version>1.7.30</version>
+            <version>1.7.32</version>
         </dependency>
         <dependency>
             <groupId>junit</groupId>
@@ -242,9 +242,9 @@
                 <artifactId>maven-compiler-plugin</artifactId>
                 <version>3.8.1</version>
                 <configuration>
-                    <release>9</release>
-                    <source>9</source>
-                    <target>9</target>
+                    <release>8</release>
+                    <source>8</source>
+                    <target>8</target>
                 </configuration>
             </plugin>
             <plugin>

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

@@ -23,11 +23,13 @@
 import com.google.common.util.concurrent.UncheckedExecutionException;
 import java.util.concurrent.ExecutionException;
 import org.xbill.DNS.Lookup;
+import org.xbill.DNS.lookup.LookupSession;
 
 /**
  * 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.
  */
+@Deprecated
 class CachingLookupFactory implements LookupFactory {
   private final LookupFactory delegate;
   private final ThreadLocal<Cache<String, Lookup>> cacheHolder;
@@ -51,4 +53,9 @@ public Lookup forName(final String fqdn) {
       throw new DnsException(e);
     }
   }
+
+  @Override
+  public LookupSession sessionForName(String fqdn) {
+    throw new java.lang.UnsupportedOperationException("Session not supported with caching lookup");
+  }
 }

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.
@@ -25,10 +26,24 @@ public interface DnsSrvResolver {
   /**
    * Does a DNS SRV lookup for the supplied fully qualified domain name, and returns the
    * matching results.
+   * @deprecated
+   * This method is deprecated in favor of the asynchronous version.
+   * Use {@link DnsSrvResolver#resolveAsync(String)} instead
    *
    * @param fqdn a DNS name to query for
    * @return a possibly empty list of matching records
    * @throws DnsException if there was an error doing the DNS lookup
    */
+  @Deprecated
   List<LookupResult> resolve(String fqdn);
+
+  /**
+   * Does a DNS SRV lookup for the supplied fully qualified domain name, and returns the
+   * matching results.
+   *
+   * @param fqdn a DNS name to query for
+   * @return a possibly empty list of matching records
+   * @throws DnsException if there was an error doing the DNS lookup
+   */
+  CompletionStage<List<LookupResult>> resolveAsync(String fqdn);
 }

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

@@ -23,6 +23,9 @@
 import java.net.UnknownHostException;
 import java.time.Duration;
 import java.util.List;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
+
 import org.xbill.DNS.ExtendedResolver;
 import org.xbill.DNS.Resolver;
 
@@ -46,13 +49,15 @@ public static final class DnsSrvResolverBuilder {
     private final long dnsLookupTimeoutMillis;
     private final long retentionDurationMillis;
     private final List<String> servers;
+    private final Executor executor;
 
     private DnsSrvResolverBuilder() {
       this(null,
            false,
            false,
            SECONDS.toMillis(DEFAULT_DNS_TIMEOUT_SECONDS),
            HOURS.toMillis(DEFAULT_RETENTION_DURATION_HOURS),
+           null,
            null);
     }
 
@@ -62,13 +67,15 @@ private DnsSrvResolverBuilder(
         boolean cacheLookups,
         long dnsLookupTimeoutMillis,
         long retentionDurationMillis,
-        List<String> servers) {
+        List<String> servers,
+        Executor executor) {
       this.reporter = reporter;
       this.retainData = retainData;
       this.cacheLookups = cacheLookups;
       this.dnsLookupTimeoutMillis = dnsLookupTimeoutMillis;
       this.retentionDurationMillis = retentionDurationMillis;
       this.servers = servers;
+      this.executor = executor;
     }
 
     public DnsSrvResolver build() {
@@ -79,7 +86,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);
       }
@@ -88,7 +95,8 @@ public DnsSrvResolver build() {
       final Duration timeoutDuration = Duration.ofMillis(dnsLookupTimeoutMillis);
       resolver.setTimeout(timeoutDuration);
 
-      LookupFactory lookupFactory = new SimpleLookupFactory(resolver);
+      LookupFactory lookupFactory = executor == null ? new SimpleLookupFactory(resolver, ForkJoinPool.commonPool()) :
+              new SimpleLookupFactory(resolver, executor);
 
       if (cacheLookups) {
         lookupFactory = new CachingLookupFactory(lookupFactory);
@@ -109,27 +117,37 @@ public DnsSrvResolver build() {
 
     public DnsSrvResolverBuilder metered(DnsReporter reporter) {
       return new DnsSrvResolverBuilder(reporter, retainData, cacheLookups, dnsLookupTimeoutMillis,
-                                       retentionDurationMillis, servers);
+                                       retentionDurationMillis, servers, executor);
     }
 
     public DnsSrvResolverBuilder retainingDataOnFailures(boolean retainData) {
       return new DnsSrvResolverBuilder(reporter, retainData, cacheLookups, dnsLookupTimeoutMillis,
-                                       retentionDurationMillis, servers);
+                                       retentionDurationMillis, servers, executor);
     }
 
+    /**
+     * @deprecated
+     * CachingLookups will be removed in the future as it doesn't work with `resolveAsync`
+     */
+    @Deprecated
     public DnsSrvResolverBuilder cachingLookups(boolean cacheLookups) {
       return new DnsSrvResolverBuilder(reporter, retainData, cacheLookups, dnsLookupTimeoutMillis,
-                                       retentionDurationMillis, servers);
+                                       retentionDurationMillis, servers, executor);
     }
 
     public DnsSrvResolverBuilder dnsLookupTimeoutMillis(long dnsLookupTimeoutMillis) {
       return new DnsSrvResolverBuilder(reporter, retainData, cacheLookups, dnsLookupTimeoutMillis,
-                                       retentionDurationMillis, servers);
+                                       retentionDurationMillis, servers, executor);
     }
 
     public DnsSrvResolverBuilder retentionDurationMillis(long retentionDurationMillis) {
       return new DnsSrvResolverBuilder(reporter, retainData, cacheLookups, dnsLookupTimeoutMillis,
-                                       retentionDurationMillis, servers);
+                                       retentionDurationMillis, servers, executor);
+    }
+
+    public DnsSrvResolverBuilder executor(Executor executor) {
+      return new DnsSrvResolverBuilder(reporter, retainData, cacheLookups, dnsLookupTimeoutMillis,
+              retentionDurationMillis, servers, executor);
     }
 
     /**
@@ -143,7 +161,7 @@ public DnsSrvResolverBuilder retentionDurationMillis(long retentionDurationMilli
      */
     public DnsSrvResolverBuilder servers(List<String> servers) {
       return new DnsSrvResolverBuilder(reporter, retainData, cacheLookups, dnsLookupTimeoutMillis,
-                                       retentionDurationMillis, servers);
+                                       retentionDurationMillis, servers, executor);
     }
   }
 

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

@@ -17,15 +17,28 @@
 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} or {@link Lookup} instances.
  */
 interface LookupFactory {
   /**
    * Returns a {@link Lookup} instance capable of doing SRV lookups for the supplied FQDN.
+   * @deprecated
+   * This synchronous method is being deprecated.
+   * Use {@link LookupFactory#sessionForName(String)} instead
+   *
    * @param fqdn the name to do lookups for
    * @return a Lookup instance
    */
+  @Deprecated
   Lookup forName(String 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
+   */
+  LookupSession sessionForName(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.
@@ -35,28 +37,54 @@ class MeteredDnsSrvResolver implements DnsSrvResolver {
     this.reporter = requireNonNull(reporter, "reporter");
   }
 
+    @Override
+    public 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;
+
+        try {
+            result = delegate.resolve(fqdn);
+        } catch (RuntimeException error) {
+            reporter.reportFailure(error);
+            throw error;
+        } finally {
+            resolveTimer.stop();
+        }
+
+        if (result.isEmpty()) {
+            reporter.reportEmpty();
+        }
+
+        return result;
+    }
+
   @Override
-  public List<LookupResult> resolve(String fqdn) {
+  public CompletionStage<List<LookupResult>> resolveAsync(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;
-
-    try {
-      result = delegate.resolve(fqdn);
-    } catch (RuntimeException error) {
-      reporter.reportFailure(error);
-      throw error;
-    } finally {
-      resolveTimer.stop();
-    }
-
-    if (result.isEmpty()) {
-      reporter.reportEmpty();
-    }
+    return delegate
+        .resolveAsync(fqdn)
+        .handle(
+            (result, error) -> {
+              resolveTimer.stop();
+              if (error == null) {
+                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;
 
 /**
@@ -70,4 +71,29 @@ public List<LookupResult> resolve(final String fqdn) {
       throw new RuntimeException(e);
     }
   }
+
+  @Override
+  public CompletionStage<List<LookupResult>> resolveAsync(final String fqdn) {
+    requireNonNull(fqdn, "fqdn");
+    return delegate.resolveAsync(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);
+      }
+    });
+  }
 }