Insecure source port usage for DNS queries in Graylog

@@ -0,0 +1,2 @@
+type = "security"
+message = "Fix insecure source port usage for DNS Lookup adapter queries. [GHSA-g96c-x7rh-99r3](https://github.com/Graylog2/graylog2-server/security/advisories/GHSA-g96c-x7rh-99r3)"

@@ -80,6 +80,7 @@
 import org.graylog2.indexer.retention.RetentionStrategyBindings;
 import org.graylog2.indexer.rotation.RotationStrategyBindings;
 import org.graylog2.inputs.transports.NettyTransportConfiguration;
+import org.graylog2.lookup.adapters.dnslookup.DnsLookupAdapterConfiguration;
 import org.graylog2.messageprocessors.MessageProcessorModule;
 import org.graylog2.migrations.MigrationsModule;
 import org.graylog2.notifications.Notification;
@@ -133,8 +134,9 @@ public class Server extends ServerBootstrap {
     private final PrometheusExporterConfiguration prometheusExporterConfiguration = new PrometheusExporterConfiguration();
     private final TLSProtocolsConfiguration tlsConfiguration = new TLSProtocolsConfiguration();
     private final GeoIpProcessorConfig geoIpProcessorConfig = new GeoIpProcessorConfig();
-
     private final TelemetryConfiguration telemetryConfiguration = new TelemetryConfiguration();
+    private final DnsLookupAdapterConfiguration dnsLookupAdapterConfiguration = new DnsLookupAdapterConfiguration();
+
     @Option(name = {"-l", "--local"}, description = "Run Graylog in local mode. Only interesting for Graylog developers.")
     private boolean local = false;
 
@@ -217,7 +219,8 @@ protected List<Object> getCommandConfigurationBeans() {
                 prometheusExporterConfiguration,
                 tlsConfiguration,
                 geoIpProcessorConfig,
-                telemetryConfiguration);
+                telemetryConfiguration,
+                dnsLookupAdapterConfiguration);
     }
 
     @Override

@@ -37,6 +37,7 @@
 import org.graylog2.lookup.adapters.dnslookup.ADnsAnswer;
 import org.graylog2.lookup.adapters.dnslookup.DnsAnswer;
 import org.graylog2.lookup.adapters.dnslookup.DnsClient;
+import org.graylog2.lookup.adapters.dnslookup.DnsLookupAdapterConfiguration;
 import org.graylog2.lookup.adapters.dnslookup.DnsLookupType;
 import org.graylog2.lookup.adapters.dnslookup.PtrDnsAnswer;
 import org.graylog2.lookup.adapters.dnslookup.TxtDnsAnswer;
@@ -80,6 +81,7 @@ public class DnsLookupDataAdapter extends LookupDataAdapter {
     private static final String TIMER_TEXT_LOOKUP = "textLookupTime";
     private DnsClient dnsClient;
     private final Config config;
+    private final DnsLookupAdapterConfiguration adapterConfiguration;
 
     private final Counter errorCounter;
 
@@ -90,9 +92,11 @@ public class DnsLookupDataAdapter extends LookupDataAdapter {
 
     @Inject
     public DnsLookupDataAdapter(@Assisted("dto") DataAdapterDto dto,
-                                MetricRegistry metricRegistry) {
+                                MetricRegistry metricRegistry,
+                                DnsLookupAdapterConfiguration adapterConfiguration) {
         super(dto, metricRegistry);
         this.config = (Config) dto.config();
+        this.adapterConfiguration = adapterConfiguration;
         this.errorCounter = metricRegistry.counter(MetricRegistry.name(getClass(), dto.id(), ERROR_COUNTER));
         this.resolveDomainNameTimer = metricRegistry.timer(MetricRegistry.name(getClass(), dto.id(), TIMER_RESOLVE_DOMAIN_NAME));
         this.reverseLookupTimer = metricRegistry.timer(MetricRegistry.name(getClass(), dto.id(), TIMER_REVERSE_LOOKUP));
@@ -101,8 +105,8 @@ public DnsLookupDataAdapter(@Assisted("dto") DataAdapterDto dto,
 
     @Override
     protected void doStart() {
-
-        dnsClient = new DnsClient(config.requestTimeout());
+        dnsClient = new DnsClient(config.requestTimeout(), adapterConfiguration.getPoolSize(),
+                adapterConfiguration.getPoolRefreshInterval().toSeconds());
         dnsClient.start(config.serverIps());
     }
 

@@ -24,8 +24,6 @@
 import com.google.common.net.InetAddresses;
 import com.google.common.net.InternetDomainName;
 import io.netty.buffer.ByteBuf;
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.channel.socket.nio.NioDatagramChannel;
 import io.netty.handler.codec.dns.DefaultDnsPtrRecord;
 import io.netty.handler.codec.dns.DefaultDnsQuestion;
 import io.netty.handler.codec.dns.DefaultDnsRawRecord;
@@ -35,20 +33,15 @@
 import io.netty.handler.codec.dns.DnsResponse;
 import io.netty.handler.codec.dns.DnsSection;
 import io.netty.resolver.dns.DnsNameResolver;
-import io.netty.resolver.dns.DnsNameResolverBuilder;
-import io.netty.resolver.dns.DnsServerAddressStreamProvider;
-import io.netty.resolver.dns.SequentialDnsServerAddressStreamProvider;
-import io.netty.util.concurrent.Future;
-import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.graylog2.lookup.adapters.dnslookup.DnsResolverPool.ResolverLease;
 import org.graylog2.shared.utilities.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.net.Inet4Address;
 import java.net.Inet6Address;
 import java.net.InetAddress;
-import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.List;
@@ -59,10 +52,11 @@
 import java.util.concurrent.TimeoutException;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
-import java.util.stream.StreamSupport;
 
-public class DnsClient {
+import static org.graylog2.lookup.adapters.dnslookup.DnsLookupAdapterConfiguration.DEFAULT_POOL_SIZE;
+import static org.graylog2.lookup.adapters.dnslookup.DnsLookupAdapterConfiguration.DEFAULT_REFRESH_INTERVAL_SECONDS;
 
+public class DnsClient {
     private static final Logger LOG = LoggerFactory.getLogger(DnsClient.class);
 
     private static final int DEFAULT_DNS_PORT = 53;
@@ -80,9 +74,9 @@ public class DnsClient {
     private static final char[] HEX_CHARS_ARRAY = "0123456789ABCDEF".toCharArray();
     private final long queryTimeout;
     private final long requestTimeout;
-
-    private NioEventLoopGroup nettyEventLoop;
-    private DnsNameResolver resolver;
+    private final long resolverPoolSize;
+    private final long resolverPoolRefreshSeconds;
+    private DnsResolverPool resolverPool;
 
     /**
      * Creates a new DNS client with the given query timeout. The request timeout will be the query timeout plus
@@ -108,66 +102,33 @@ public DnsClient(long queryTimeout) {
      * @param requestTimeout the request timeout
      */
     public DnsClient(long queryTimeout, long requestTimeout) {
+        this(queryTimeout, requestTimeout, DEFAULT_POOL_SIZE, DEFAULT_REFRESH_INTERVAL_SECONDS);
+    }
+
+    public DnsClient(long queryTimeout, int resolverPoolSize, long resolverPoolRefreshSeconds) {
+        this(queryTimeout, queryTimeout + DEFAULT_REQUEST_TIMEOUT_INCREMENT, resolverPoolSize, resolverPoolRefreshSeconds);
+    }
+
+    private DnsClient(long queryTimeout, long requestTimeout, int resolverPoolSize, long resolverPoolRefreshSeconds) {
         this.queryTimeout = queryTimeout;
         this.requestTimeout = requestTimeout;
+        this.resolverPoolSize = resolverPoolSize;
+        this.resolverPoolRefreshSeconds = resolverPoolRefreshSeconds;
     }
 
     public void start(String dnsServerIps) {
-
         LOG.debug("Attempting to start DNS client");
-        final List<InetSocketAddress> iNetDnsServerIps = parseServerIpAddresses(dnsServerIps);
-
-        nettyEventLoop = new NioEventLoopGroup();
-
-        final DnsNameResolverBuilder dnsNameResolverBuilder = new DnsNameResolverBuilder(nettyEventLoop.next());
-        dnsNameResolverBuilder.channelType(NioDatagramChannel.class).queryTimeoutMillis(queryTimeout);
-
-        // Specify custom DNS servers if provided. If not, use those specified in local network adapter settings.
-        if (CollectionUtils.isNotEmpty(iNetDnsServerIps)) {
-
-            LOG.debug("Attempting to start DNS client with server IPs [{}] on port [{}] with timeout [{}]",
-                      dnsServerIps, DEFAULT_DNS_PORT, requestTimeout);
-
-            final DnsServerAddressStreamProvider dnsServer = new SequentialDnsServerAddressStreamProvider(iNetDnsServerIps);
-            dnsNameResolverBuilder.nameServerProvider(dnsServer);
-        } else {
-            LOG.debug("Attempting to start DNS client with the local network adapter DNS server address on port [{}] with timeout [{}]",
-                      DEFAULT_DNS_PORT, requestTimeout);
-        }
-
-        resolver = dnsNameResolverBuilder.build();
-
-        LOG.debug("DNS client startup successful");
-    }
-
-    private List<InetSocketAddress> parseServerIpAddresses(String dnsServerIps) {
-
-        // Parse and prepare DNS server IP addresses for Netty.
-        return StreamSupport
-                // Split comma-separated sever IP:port combos.
-                .stream(Splitter.on(",").trimResults().omitEmptyStrings().split(dnsServerIps).spliterator(), false)
-                // Parse as HostAndPort objects (allows convenient handling of port provided after colon).
-                .map(hostAndPort -> HostAndPort.fromString(hostAndPort).withDefaultPort(DnsClient.DEFAULT_DNS_PORT))
-                // Convert HostAndPort > InetSocketAddress as required by Netty.
-                .map(hostAndPort -> new InetSocketAddress(hostAndPort.getHost(), hostAndPort.getPort()))
-                .collect(Collectors.toList());
+        this.resolverPool = new DnsResolverPool(dnsServerIps, queryTimeout, resolverPoolSize, resolverPoolRefreshSeconds);
+        this.resolverPool.initialize();
     }
 
     public void stop() {
-
         LOG.debug("Attempting to stop DNS client");
-
-        if (nettyEventLoop == null) {
-            LOG.error("DNS resolution event loop not initialized");
+        if (resolverPool == null) {
+            LOG.error("DNS resolution pool is not initialized.");
             return;
         }
-
-        // Make sure to close the resolver before shutting down the event loop
-        resolver.close();
-
-        // Shutdown event loop (required by Netty).
-        final Future<?> shutdownFuture = nettyEventLoop.shutdownGracefully();
-        shutdownFuture.addListener(future -> LOG.debug("DNS client shutdown successful"));
+        resolverPool.stop();
     }
 
     public List<ADnsAnswer> resolveIPv4AddressForHostname(String hostName, boolean includeIpVersion)
@@ -187,24 +148,28 @@ private List<ADnsAnswer> resolveIpAddresses(String hostName, DnsRecordType dnsRe
 
         LOG.debug("Attempting to resolve [{}] records for [{}]", dnsRecordType, hostName);
 
-        if (isShutdown()) {
+        if (resolverPool.isStopped()) {
             throw new DnsClientNotRunningException();
         }
 
         validateHostName(hostName);
 
         final DefaultDnsQuestion aRecordDnsQuestion = new DefaultDnsQuestion(hostName, dnsRecordType);
 
+        final ResolverLease resolverLease = resolverPool.takeLease();
         /* The DnsNameResolver.resolveAll(DnsQuestion) method handles all redirects through CNAME records to
          * ultimately resolve a list of IP addresses with TTL values. */
         try {
-            return resolver.resolveAll(aRecordDnsQuestion).get(requestTimeout, TimeUnit.MILLISECONDS).stream()
+            return resolverLease.getResolver().resolveAll(aRecordDnsQuestion).get(requestTimeout, TimeUnit.MILLISECONDS).stream()
                            .map(dnsRecord -> decodeDnsRecord(dnsRecord, includeIpVersion))
                            .filter(Objects::nonNull) // Removes any entries which the IP address could not be extracted for.
                            .collect(Collectors.toList());
         } catch (TimeoutException e) {
             throw new ExecutionException("Resolver future didn't return a result in " + requestTimeout + " ms", e);
         }
+        finally {
+            resolverPool.returnLease(resolverLease);
+        }
     }
 
     /**
@@ -262,7 +227,7 @@ public PtrDnsAnswer reverseLookup(String ipAddress) throws InterruptedException,
 
         LOG.debug("Attempting to perform reverse lookup for IP address [{}]", ipAddress);
 
-        if (isShutdown()) {
+        if (resolverPool.isStopped()) {
             throw new DnsClientNotRunningException();
         }
 
@@ -271,8 +236,9 @@ public PtrDnsAnswer reverseLookup(String ipAddress) throws InterruptedException,
         final String inverseAddressFormat = getInverseAddressFormat(ipAddress);
 
         DnsResponse content = null;
+        final ResolverLease resolverLease = resolverPool.takeLease();
         try {
-            content = resolver.query(new DefaultDnsQuestion(inverseAddressFormat, DnsRecordType.PTR)).get(requestTimeout, TimeUnit.MILLISECONDS).content();
+            content = resolverLease.getResolver().query(new DefaultDnsQuestion(inverseAddressFormat, DnsRecordType.PTR)).get(requestTimeout, TimeUnit.MILLISECONDS).content();
             for (int i = 0; i < content.count(DnsSection.ANSWER); i++) {
 
                 // Return the first PTR record, because there should be only one as per
@@ -306,6 +272,7 @@ public PtrDnsAnswer reverseLookup(String ipAddress) throws InterruptedException,
                 // Must manually release references on content object since the DnsResponse class extends ReferenceCounted
                 content.release();
             }
+            resolverPool.returnLease(resolverLease);
         }
 
         return null;
@@ -348,7 +315,7 @@ public static void parseReverseLookupDomain(PtrDnsAnswer.Builder dnsAnswerBuilde
 
     public List<TxtDnsAnswer> txtLookup(String hostName) throws InterruptedException, ExecutionException {
 
-        if (isShutdown()) {
+        if (resolverPool.isStopped()) {
             throw new DnsClientNotRunningException();
         }
 
@@ -357,8 +324,9 @@ public List<TxtDnsAnswer> txtLookup(String hostName) throws InterruptedException
         validateHostName(hostName);
 
         DnsResponse content = null;
+        final ResolverLease resolverLease = resolverPool.takeLease();
         try {
-            content = resolver.query(new DefaultDnsQuestion(hostName, DnsRecordType.TXT)).get(requestTimeout, TimeUnit.MILLISECONDS).content();
+            content = resolverLease.getResolver().query(new DefaultDnsQuestion(hostName, DnsRecordType.TXT)).get(requestTimeout, TimeUnit.MILLISECONDS).content();
             int count = content.count(DnsSection.ANSWER);
             final ArrayList<TxtDnsAnswer> txtRecords = new ArrayList<>(count);
             for (int i = 0; i < count; i++) {
@@ -389,13 +357,10 @@ public List<TxtDnsAnswer> txtLookup(String hostName) throws InterruptedException
                 // Must manually release references on content object since the DnsResponse class extends ReferenceCounted
                 content.release();
             }
+            resolverPool.returnLease(resolverLease);
         }
     }
 
-    private boolean isShutdown() {
-        return nettyEventLoop == null || nettyEventLoop.isShutdown();
-    }
-
     private static String decodeTxtRecord(DefaultDnsRawRecord record) {
 
         LOG.debug("Attempting to read TXT value from DNS record [{}]", record);

@@ -0,0 +1,46 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog2.lookup.adapters.dnslookup;
+
+import com.github.joschi.jadconfig.Parameter;
+import com.github.joschi.jadconfig.util.Duration;
+import com.github.joschi.jadconfig.validators.PositiveDurationValidator;
+import com.github.joschi.jadconfig.validators.PositiveIntegerValidator;
+import org.graylog2.plugin.PluginConfigBean;
+
+public class DnsLookupAdapterConfiguration implements PluginConfigBean {
+    private static final String PREFIX = "dns_lookup_adapter_";
+    protected static final String RESOLVER_POOL_SIZE = PREFIX + "resolver_pool_size";
+    protected static final String RESOLVER_POOL_REFRESH_INTERVAL = PREFIX + "resolver_pool_refresh_interval";
+
+    protected static final int DEFAULT_POOL_SIZE = 10;
+    protected static final int DEFAULT_REFRESH_INTERVAL_SECONDS = 300;
+
+    @Parameter(value = RESOLVER_POOL_SIZE, validators = PositiveIntegerValidator.class)
+    private int poolSize = DEFAULT_POOL_SIZE;
+
+    @Parameter(value = RESOLVER_POOL_REFRESH_INTERVAL, validators = PositiveDurationValidator.class)
+    private Duration poolRefreshInterval = Duration.seconds(DEFAULT_REFRESH_INTERVAL_SECONDS);
+
+    public int getPoolSize() {
+        return poolSize;
+    }
+
+    public Duration getPoolRefreshInterval() {
+        return poolRefreshInterval;
+    }
+}

@@ -0,0 +1,82 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog2.lookup.adapters.dnslookup;
+
+import com.google.common.base.Splitter;
+import com.google.common.net.HostAndPort;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioDatagramChannel;
+import io.netty.resolver.dns.DnsNameResolver;
+import io.netty.resolver.dns.DnsNameResolverBuilder;
+import io.netty.resolver.dns.DnsServerAddressStreamProvider;
+import io.netty.resolver.dns.SequentialDnsServerAddressStreamProvider;
+import org.apache.commons.collections4.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+public class DnsNameResolverFactory {
+    private static final Logger LOG = LoggerFactory.getLogger(DnsNameResolverFactory.class);
+    private static final int DEFAULT_DNS_PORT = 53;
+
+    private final NioEventLoopGroup eventLoopGroup;
+    private String dnsServerIps;
+    private final long queryTimeout;
+
+    public DnsNameResolverFactory(NioEventLoopGroup eventLoopGroup, String dnsServerIps, long queryTimeout) {
+        this.eventLoopGroup = eventLoopGroup;
+        this.dnsServerIps = dnsServerIps;
+        this.queryTimeout = queryTimeout;
+    }
+
+    public DnsNameResolver create() {
+        final List<InetSocketAddress> iNetDnsServerIps = parseServerIpAddresses(dnsServerIps);
+        final DnsNameResolverBuilder dnsNameResolverBuilder = new DnsNameResolverBuilder(eventLoopGroup.next());
+        dnsNameResolverBuilder.channelType(NioDatagramChannel.class).queryTimeoutMillis(queryTimeout);
+
+        // Specify custom DNS servers if provided. If not, use those specified in local network adapter settings.
+        if (CollectionUtils.isNotEmpty(iNetDnsServerIps)) {
+            LOG.debug("Attempting to start DNS client with server IPs [{}] on port [{}].",
+                    dnsServerIps, DEFAULT_DNS_PORT);
+
+            final DnsServerAddressStreamProvider dnsServer = new SequentialDnsServerAddressStreamProvider(iNetDnsServerIps);
+            dnsNameResolverBuilder.nameServerProvider(dnsServer);
+        } else {
+            LOG.debug("Attempting to start DNS client with custom server IPs [{}] on port [{}].",
+                    dnsServerIps, DEFAULT_DNS_PORT);
+        }
+
+        return dnsNameResolverBuilder.build();
+    }
+
+    private List<InetSocketAddress> parseServerIpAddresses(String dnsServerIps) {
+
+        // Parse and prepare DNS server IP addresses for Netty.
+        return StreamSupport
+                // Split comma-separated sever IP:port combos.
+                .stream(Splitter.on(",").trimResults().omitEmptyStrings().split(dnsServerIps).spliterator(), false)
+                // Parse as HostAndPort objects (allows convenient handling of port provided after colon).
+                .map(hostAndPort -> HostAndPort.fromString(hostAndPort).withDefaultPort(DEFAULT_DNS_PORT))
+                // Convert HostAndPort > InetSocketAddress as required by Netty.
+                .map(hostAndPort -> new InetSocketAddress(hostAndPort.getHost(), hostAndPort.getPort()))
+                .collect(Collectors.toList());
+    }
+}

@@ -0,0 +1,242 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog2.lookup.adapters.dnslookup;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.resolver.dns.DnsNameResolver;
+import io.netty.util.concurrent.Future;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+/**
+ * Manages a pool of Netty {@link DnsNameResolverFactory} objects.
+ *
+ * <br>
+ * Since the source port for DNS resolution requests is fixed for the duration of the resolver lifecycle,
+ * the pooling capability of this class allows the source address to by varied for each request by choosing
+ * a random resolver from the pool for each request.
+ *
+ * <br>
+ * The resolvers in the pool are periodically refreshed to cycle in new source ports for subsequent requests.
+ *
+ * <br>
+ * The pool size and refresh interval are configurable globally for all DNS Lookup adapters with the following
+ * Graylog server configuration properties (the defaults are indicated below as well).
+ * <br>
+ * <pre>
+ * dns_lookup_adapter_resolver_pool_size = 10
+ * dns_lookup_adapter_resolver_pool_refresh_interval = 300s
+ * </pre>
+ *
+ * <br>
+ * Callers can use the {@link #takeLease()} method to acquire a lease for a resolver.
+ * The {@link ResolverLease#release()} method must be called to release a resolver lease after use.
+ * These operations are thread-safe.
+ */
+public class DnsResolverPool {
+    private static final Logger LOG = LoggerFactory.getLogger(DnsResolverPool.class);
+    private final long poolSize;
+    private final long poolRefreshSeconds;
+    private final ScheduledExecutorService executorService;
+    private final NioEventLoopGroup eventLoopGroup;
+    private final DnsNameResolverFactory resolverFactory;
+
+    // Pool is accessed by resolution requests and by refresh tasks.
+    private final List<ResolverLease> resolverPool;
+
+    protected DnsResolverPool(String dnsServerIps, long queryTimeout, long poolSize, long poolRefreshSeconds) {
+        this.poolSize = poolSize;
+        this.poolRefreshSeconds = poolRefreshSeconds;
+        this.executorService = Executors.newSingleThreadScheduledExecutor(
+                new ThreadFactoryBuilder().setNameFormat("dns-lookup-refresh-task-%d").build());
+
+        // A synchronized list is used to ensure thread safety for list mutations and accesses.
+        this.resolverPool = Collections.synchronizedList(new ArrayList<>());
+
+        // Use a single Netty EventLoopGroup (1 thread by default) for all resolvers in the pool.
+        // We are expecting DNS resolution requests for a specific pooler to be executed sequentially
+        // (one after the other) amongst the pool of resolvers. So, a single thread should be sufficient.
+        this.eventLoopGroup = new NioEventLoopGroup();
+        this.resolverFactory = new DnsNameResolverFactory(eventLoopGroup, dnsServerIps, queryTimeout);
+    }
+
+    protected void initialize() {
+        for (int i = 0; i < poolSize; i++) {
+            resolverPool.add(new ResolverLease(resolverFactory.create()));
+        }
+        executorService.scheduleAtFixedRate(new ResolverRefreshTask(), poolRefreshSeconds, poolRefreshSeconds, TimeUnit.SECONDS);
+    }
+
+    protected ResolverLease takeLease() {
+        if (resolverPool.size() == 0) {
+            throw new RuntimeException("Resolver pool is empty. Cannot return lease.");
+        }
+        final ResolverLease lease = resolverPool.get(randomResolverIndex());
+        lease.take();
+        return lease;
+    }
+
+    protected void returnLease(ResolverLease lease) {
+        lease.release();
+    }
+
+    public void stop() {
+        LOG.debug("Attempting to stop pool.");
+        executorService.shutdown();
+        if (resolverPool == null) {
+            LOG.error("Resolver pool has not been initialized.");
+            return;
+        }
+
+        synchronized (resolverPool) {
+            final Iterator<ResolverLease> iterator = resolverPool.iterator();
+            while (iterator.hasNext()) {
+                ResolverLease lease = iterator.next();
+                LOG.debug("Attempting to stop resolver [{}].", lease.getId());
+                if (lease.isLeased()) {
+                    LOG.warn("Attempting to stop a leased resolver...");
+                }
+                lease.take();
+                lease.getResolver().close();
+                iterator.remove();
+                LOG.debug("Successfully stopped resolver [{}].", lease.getId());
+            }
+        }
+
+        // Shutdown event loop (required by Netty).
+        final Future<?> shutdownFuture = eventLoopGroup.shutdownGracefully();
+        shutdownFuture.addListener(future -> LOG.debug("Finished shutting down pool."));
+        LOG.debug("Resolver pool shutdown complete.");
+    }
+
+    protected boolean isStopped() {
+        return (eventLoopGroup == null || eventLoopGroup.isShutdown()) && executorService.isShutdown();
+    }
+
+    /**
+     * Allows for a random resolver to be returned for each request.
+     */
+    protected int randomResolverIndex() {
+        // Use ThreadLocalRandom to ensure that different random numbers are returned when queried from different
+        // threads referencing the same instance.
+        return ThreadLocalRandom.current().nextInt(resolverPool.size());
+
+    }
+
+    private class ResolverRefreshTask implements Runnable {
+        @Override
+        public void run() {
+            LOG.debug("Starting resolver refresh.");
+            LOG.debug("Existing IDs: [{}]", resolverPool.stream().map(ResolverLease::getId).collect(Collectors.joining(", ")));
+            synchronized (resolverPool) {
+                final ListIterator<ResolverLease> iterator = resolverPool.listIterator();
+                while (iterator.hasNext()) {
+                    ResolverLease lease = iterator.next();
+                    if (!lease.getHasBeenLeased()) {
+                        LOG.debug("Resolver [{}] has not been leased yet. Skipping refresh.", lease.getId());
+                        continue;
+                    }
+
+                    if (!lease.isLeased()) {
+                        lease.getResolver().close();
+                        iterator.remove();
+                        iterator.add(new ResolverLease(resolverFactory.create()));
+                    } else {
+                        LOG.warn("Lease for resolver [{}] is in-use. Skipping refresh. This will be attempted again in [{}] seconds. " +
+                                        "If this happens frequently for high message rates, consider increasing the [dns_lookup_adapter_resolver_pool_size = 10] " +
+                                        "server configuration property to allow more DNS resolvers.",
+                                lease.getId(), poolRefreshSeconds);
+                    }
+                }
+            }
+            LOG.debug("Resolver IDs refreshed: [{}]", resolverPool.stream().map(ResolverLease::getId).collect(Collectors.joining(", ")));
+            LOG.debug("Finished resolver refresh.");
+        }
+    }
+
+    protected int poolSize() {
+        return resolverPool != null ? resolverPool.size() : 0;
+    }
+
+    protected static class ResolverLease {
+        private final String id;
+        private final DnsNameResolver resolver;
+        private AtomicInteger leaseCount;
+        private AtomicBoolean hasBeenLeased;
+
+        private ResolverLease(DnsNameResolver resolver) {
+            this.id = UUID.randomUUID().toString();
+            this.resolver = resolver;
+            this.leaseCount = new AtomicInteger(0);
+            this.hasBeenLeased = new AtomicBoolean();
+        }
+
+        private void take() {
+            this.leaseCount.incrementAndGet();
+            this.hasBeenLeased.set(true);
+        }
+
+        private void release() {
+            this.leaseCount.decrementAndGet();
+        }
+
+        protected String getId() {
+            return id;
+        }
+
+        private boolean isLeased() {
+            return leaseCount.get() > 0;
+        }
+
+        private boolean getHasBeenLeased() {
+            return hasBeenLeased.get();
+        }
+
+        protected DnsNameResolver getResolver() {
+            return resolver;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            final ResolverLease that = (ResolverLease) o;
+            return Objects.equals(id, that.id);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(id);
+        }
+    }
+}

@@ -0,0 +1,40 @@
+package org.graylog2.lookup.adapters.dnslookup;
+
+import org.graylog2.lookup.adapters.dnslookup.DnsResolverPool.ResolverLease;
+import org.junit.jupiter.api.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@RunWith(MockitoJUnitRunner.class)
+class DnsResolverPoolTest {
+
+    private static final int POOL_SIZE = 10;
+
+    @Test
+    public void testResolverIndex() {
+        final DnsResolverPool dnsResolverPool = new DnsResolverPool("", 100, POOL_SIZE, 300);
+        dnsResolverPool.initialize();
+        assertEquals(POOL_SIZE, dnsResolverPool.poolSize());
+        final int randomResolverIndex = dnsResolverPool.randomResolverIndex();
+        assertTrue(randomResolverIndex >= 0);
+        assertTrue(randomResolverIndex < POOL_SIZE);
+        dnsResolverPool.stop();
+    }
+
+    @Test
+    public void verifyRefresh() throws InterruptedException {
+        final DnsResolverPool dnsResolverPool = new DnsResolverPool("", 100, 1, 1);
+        dnsResolverPool.initialize();
+        final ResolverLease nextLease = dnsResolverPool.takeLease();
+        final String initialLeaseId = nextLease.getId();
+        dnsResolverPool.returnLease(nextLease);
+        Thread.sleep(1500);
+        final ResolverLease finalLease = dnsResolverPool.takeLease();
+        assertNotEquals(initialLeaseId, finalLease.getId());
+        dnsResolverPool.stop();
+    }
+}

@@ -0,0 +1,2 @@
+type = "security"
+message = "Fix insecure source port usage for DNS Lookup adapter queries. [GHSA-g96c-x7rh-99r3](https://github.com/Graylog2/graylog2-server/security/advisories/GHSA-g96c-x7rh-99r3)"

@@ -78,6 +78,7 @@
 import org.graylog2.indexer.retention.RetentionStrategyBindings;
 import org.graylog2.indexer.rotation.RotationStrategyBindings;
 import org.graylog2.inputs.transports.NettyTransportConfiguration;
+import org.graylog2.lookup.adapters.dnslookup.DnsLookupAdapterConfiguration;
 import org.graylog2.messageprocessors.MessageProcessorModule;
 import org.graylog2.migrations.MigrationsModule;
 import org.graylog2.notifications.Notification;
@@ -131,6 +132,7 @@ public class Server extends ServerBootstrap {
     private final PrometheusExporterConfiguration prometheusExporterConfiguration = new PrometheusExporterConfiguration();
     private final TLSProtocolsConfiguration tlsConfiguration = new TLSProtocolsConfiguration();
     private final GeoIpProcessorConfig geoIpProcessorConfig = new GeoIpProcessorConfig();
+    private final DnsLookupAdapterConfiguration dnsLookupAdapterConfiguration = new DnsLookupAdapterConfiguration();
 
     public Server() {
         super("server", configuration);
@@ -211,7 +213,8 @@ protected List<Object> getCommandConfigurationBeans() {
                 jobSchedulerConfiguration,
                 prometheusExporterConfiguration,
                 tlsConfiguration,
-                geoIpProcessorConfig);
+                geoIpProcessorConfig,
+                dnsLookupAdapterConfiguration);
     }
 
     @Override

@@ -37,6 +37,7 @@
 import org.graylog2.lookup.adapters.dnslookup.ADnsAnswer;
 import org.graylog2.lookup.adapters.dnslookup.DnsAnswer;
 import org.graylog2.lookup.adapters.dnslookup.DnsClient;
+import org.graylog2.lookup.adapters.dnslookup.DnsLookupAdapterConfiguration;
 import org.graylog2.lookup.adapters.dnslookup.DnsLookupType;
 import org.graylog2.lookup.adapters.dnslookup.PtrDnsAnswer;
 import org.graylog2.lookup.adapters.dnslookup.TxtDnsAnswer;
@@ -80,6 +81,7 @@ public class DnsLookupDataAdapter extends LookupDataAdapter {
     private static final String TIMER_TEXT_LOOKUP = "textLookupTime";
     private DnsClient dnsClient;
     private final Config config;
+    private final DnsLookupAdapterConfiguration adapterConfiguration;
 
     private final Counter errorCounter;
 
@@ -90,9 +92,11 @@ public class DnsLookupDataAdapter extends LookupDataAdapter {
 
     @Inject
     public DnsLookupDataAdapter(@Assisted("dto") DataAdapterDto dto,
-                                MetricRegistry metricRegistry) {
+                                MetricRegistry metricRegistry,
+                                DnsLookupAdapterConfiguration adapterConfiguration) {
         super(dto, metricRegistry);
         this.config = (Config) dto.config();
+        this.adapterConfiguration = adapterConfiguration;
         this.errorCounter = metricRegistry.counter(MetricRegistry.name(getClass(), dto.id(), ERROR_COUNTER));
         this.resolveDomainNameTimer = metricRegistry.timer(MetricRegistry.name(getClass(), dto.id(), TIMER_RESOLVE_DOMAIN_NAME));
         this.reverseLookupTimer = metricRegistry.timer(MetricRegistry.name(getClass(), dto.id(), TIMER_REVERSE_LOOKUP));
@@ -101,8 +105,8 @@ public DnsLookupDataAdapter(@Assisted("dto") DataAdapterDto dto,
 
     @Override
     protected void doStart() {
-
-        dnsClient = new DnsClient(config.requestTimeout());
+        dnsClient = new DnsClient(config.requestTimeout(), adapterConfiguration.getPoolSize(),
+                adapterConfiguration.getPoolRefreshInterval().toSeconds());
         dnsClient.start(config.serverIps());
     }
 

@@ -24,8 +24,6 @@
 import com.google.common.net.InetAddresses;
 import com.google.common.net.InternetDomainName;
 import io.netty.buffer.ByteBuf;
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.channel.socket.nio.NioDatagramChannel;
 import io.netty.handler.codec.dns.DefaultDnsPtrRecord;
 import io.netty.handler.codec.dns.DefaultDnsQuestion;
 import io.netty.handler.codec.dns.DefaultDnsRawRecord;
@@ -35,20 +33,15 @@
 import io.netty.handler.codec.dns.DnsResponse;
 import io.netty.handler.codec.dns.DnsSection;
 import io.netty.resolver.dns.DnsNameResolver;
-import io.netty.resolver.dns.DnsNameResolverBuilder;
-import io.netty.resolver.dns.DnsServerAddressStreamProvider;
-import io.netty.resolver.dns.SequentialDnsServerAddressStreamProvider;
-import io.netty.util.concurrent.Future;
-import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.graylog2.lookup.adapters.dnslookup.DnsResolverPool.ResolverLease;
 import org.graylog2.shared.utilities.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.net.Inet4Address;
 import java.net.Inet6Address;
 import java.net.InetAddress;
-import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.List;
@@ -59,10 +52,11 @@
 import java.util.concurrent.TimeoutException;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
-import java.util.stream.StreamSupport;
 
-public class DnsClient {
+import static org.graylog2.lookup.adapters.dnslookup.DnsLookupAdapterConfiguration.DEFAULT_POOL_SIZE;
+import static org.graylog2.lookup.adapters.dnslookup.DnsLookupAdapterConfiguration.DEFAULT_REFRESH_INTERVAL_SECONDS;
 
+public class DnsClient {
     private static final Logger LOG = LoggerFactory.getLogger(DnsClient.class);
 
     private static final int DEFAULT_DNS_PORT = 53;
@@ -80,9 +74,9 @@ public class DnsClient {
     private static final char[] HEX_CHARS_ARRAY = "0123456789ABCDEF".toCharArray();
     private final long queryTimeout;
     private final long requestTimeout;
-
-    private NioEventLoopGroup nettyEventLoop;
-    private DnsNameResolver resolver;
+    private final long resolverPoolSize;
+    private final long resolverPoolRefreshSeconds;
+    private DnsResolverPool resolverPool;
 
     /**
      * Creates a new DNS client with the given query timeout. The request timeout will be the query timeout plus
@@ -108,66 +102,33 @@ public DnsClient(long queryTimeout) {
      * @param requestTimeout the request timeout
      */
     public DnsClient(long queryTimeout, long requestTimeout) {
+        this(queryTimeout, requestTimeout, DEFAULT_POOL_SIZE, DEFAULT_REFRESH_INTERVAL_SECONDS);
+    }
+
+    public DnsClient(long queryTimeout, int resolverPoolSize, long resolverPoolRefreshSeconds) {
+        this(queryTimeout, queryTimeout + DEFAULT_REQUEST_TIMEOUT_INCREMENT, resolverPoolSize, resolverPoolRefreshSeconds);
+    }
+
+    private DnsClient(long queryTimeout, long requestTimeout, int resolverPoolSize, long resolverPoolRefreshSeconds) {
         this.queryTimeout = queryTimeout;
         this.requestTimeout = requestTimeout;
+        this.resolverPoolSize = resolverPoolSize;
+        this.resolverPoolRefreshSeconds = resolverPoolRefreshSeconds;
     }
 
     public void start(String dnsServerIps) {
-
         LOG.debug("Attempting to start DNS client");
-        final List<InetSocketAddress> iNetDnsServerIps = parseServerIpAddresses(dnsServerIps);
-
-        nettyEventLoop = new NioEventLoopGroup();
-
-        final DnsNameResolverBuilder dnsNameResolverBuilder = new DnsNameResolverBuilder(nettyEventLoop.next());
-        dnsNameResolverBuilder.channelType(NioDatagramChannel.class).queryTimeoutMillis(queryTimeout);
-
-        // Specify custom DNS servers if provided. If not, use those specified in local network adapter settings.
-        if (CollectionUtils.isNotEmpty(iNetDnsServerIps)) {
-
-            LOG.debug("Attempting to start DNS client with server IPs [{}] on port [{}] with timeout [{}]",
-                      dnsServerIps, DEFAULT_DNS_PORT, requestTimeout);
-
-            final DnsServerAddressStreamProvider dnsServer = new SequentialDnsServerAddressStreamProvider(iNetDnsServerIps);
-            dnsNameResolverBuilder.nameServerProvider(dnsServer);
-        } else {
-            LOG.debug("Attempting to start DNS client with the local network adapter DNS server address on port [{}] with timeout [{}]",
-                      DEFAULT_DNS_PORT, requestTimeout);
-        }
-
-        resolver = dnsNameResolverBuilder.build();
-
-        LOG.debug("DNS client startup successful");
-    }
-
-    private List<InetSocketAddress> parseServerIpAddresses(String dnsServerIps) {
-
-        // Parse and prepare DNS server IP addresses for Netty.
-        return StreamSupport
-                // Split comma-separated sever IP:port combos.
-                .stream(Splitter.on(",").trimResults().omitEmptyStrings().split(dnsServerIps).spliterator(), false)
-                // Parse as HostAndPort objects (allows convenient handling of port provided after colon).
-                .map(hostAndPort -> HostAndPort.fromString(hostAndPort).withDefaultPort(DnsClient.DEFAULT_DNS_PORT))
-                // Convert HostAndPort > InetSocketAddress as required by Netty.
-                .map(hostAndPort -> new InetSocketAddress(hostAndPort.getHost(), hostAndPort.getPort()))
-                .collect(Collectors.toList());
+        this.resolverPool = new DnsResolverPool(dnsServerIps, queryTimeout, resolverPoolSize, resolverPoolRefreshSeconds);
+        this.resolverPool.initialize();
     }
 
     public void stop() {
-
         LOG.debug("Attempting to stop DNS client");
-
-        if (nettyEventLoop == null) {
-            LOG.error("DNS resolution event loop not initialized");
+        if (resolverPool == null) {
+            LOG.error("DNS resolution pool is not initialized.");
             return;
         }
-
-        // Make sure to close the resolver before shutting down the event loop
-        resolver.close();
-
-        // Shutdown event loop (required by Netty).
-        final Future<?> shutdownFuture = nettyEventLoop.shutdownGracefully();
-        shutdownFuture.addListener(future -> LOG.debug("DNS client shutdown successful"));
+        resolverPool.stop();
     }
 
     public List<ADnsAnswer> resolveIPv4AddressForHostname(String hostName, boolean includeIpVersion)
@@ -187,24 +148,28 @@ private List<ADnsAnswer> resolveIpAddresses(String hostName, DnsRecordType dnsRe
 
         LOG.debug("Attempting to resolve [{}] records for [{}]", dnsRecordType, hostName);
 
-        if (isShutdown()) {
+        if (resolverPool.isStopped()) {
             throw new DnsClientNotRunningException();
         }
 
         validateHostName(hostName);
 
         final DefaultDnsQuestion aRecordDnsQuestion = new DefaultDnsQuestion(hostName, dnsRecordType);
 
+        final ResolverLease resolverLease = resolverPool.takeLease();
         /* The DnsNameResolver.resolveAll(DnsQuestion) method handles all redirects through CNAME records to
          * ultimately resolve a list of IP addresses with TTL values. */
         try {
-            return resolver.resolveAll(aRecordDnsQuestion).get(requestTimeout, TimeUnit.MILLISECONDS).stream()
+            return resolverLease.getResolver().resolveAll(aRecordDnsQuestion).get(requestTimeout, TimeUnit.MILLISECONDS).stream()
                            .map(dnsRecord -> decodeDnsRecord(dnsRecord, includeIpVersion))
                            .filter(Objects::nonNull) // Removes any entries which the IP address could not be extracted for.
                            .collect(Collectors.toList());
         } catch (TimeoutException e) {
             throw new ExecutionException("Resolver future didn't return a result in " + requestTimeout + " ms", e);
         }
+        finally {
+            resolverPool.returnLease(resolverLease);
+        }
     }
 
     /**
@@ -262,7 +227,7 @@ public PtrDnsAnswer reverseLookup(String ipAddress) throws InterruptedException,
 
         LOG.debug("Attempting to perform reverse lookup for IP address [{}]", ipAddress);
 
-        if (isShutdown()) {
+        if (resolverPool.isStopped()) {
             throw new DnsClientNotRunningException();
         }
 
@@ -271,8 +236,9 @@ public PtrDnsAnswer reverseLookup(String ipAddress) throws InterruptedException,
         final String inverseAddressFormat = getInverseAddressFormat(ipAddress);
 
         DnsResponse content = null;
+        final ResolverLease resolverLease = resolverPool.takeLease();
         try {
-            content = resolver.query(new DefaultDnsQuestion(inverseAddressFormat, DnsRecordType.PTR)).get(requestTimeout, TimeUnit.MILLISECONDS).content();
+            content = resolverLease.getResolver().query(new DefaultDnsQuestion(inverseAddressFormat, DnsRecordType.PTR)).get(requestTimeout, TimeUnit.MILLISECONDS).content();
             for (int i = 0; i < content.count(DnsSection.ANSWER); i++) {
 
                 // Return the first PTR record, because there should be only one as per
@@ -306,6 +272,7 @@ public PtrDnsAnswer reverseLookup(String ipAddress) throws InterruptedException,
                 // Must manually release references on content object since the DnsResponse class extends ReferenceCounted
                 content.release();
             }
+            resolverPool.returnLease(resolverLease);
         }
 
         return null;
@@ -348,7 +315,7 @@ public static void parseReverseLookupDomain(PtrDnsAnswer.Builder dnsAnswerBuilde
 
     public List<TxtDnsAnswer> txtLookup(String hostName) throws InterruptedException, ExecutionException {
 
-        if (isShutdown()) {
+        if (resolverPool.isStopped()) {
             throw new DnsClientNotRunningException();
         }
 
@@ -357,8 +324,9 @@ public List<TxtDnsAnswer> txtLookup(String hostName) throws InterruptedException
         validateHostName(hostName);
 
         DnsResponse content = null;
+        final ResolverLease resolverLease = resolverPool.takeLease();
         try {
-            content = resolver.query(new DefaultDnsQuestion(hostName, DnsRecordType.TXT)).get(requestTimeout, TimeUnit.MILLISECONDS).content();
+            content = resolverLease.getResolver().query(new DefaultDnsQuestion(hostName, DnsRecordType.TXT)).get(requestTimeout, TimeUnit.MILLISECONDS).content();
             int count = content.count(DnsSection.ANSWER);
             final ArrayList<TxtDnsAnswer> txtRecords = new ArrayList<>(count);
             for (int i = 0; i < count; i++) {
@@ -389,13 +357,10 @@ public List<TxtDnsAnswer> txtLookup(String hostName) throws InterruptedException
                 // Must manually release references on content object since the DnsResponse class extends ReferenceCounted
                 content.release();
             }
+            resolverPool.returnLease(resolverLease);
         }
     }
 
-    private boolean isShutdown() {
-        return nettyEventLoop == null || nettyEventLoop.isShutdown();
-    }
-
     private static String decodeTxtRecord(DefaultDnsRawRecord record) {
 
         LOG.debug("Attempting to read TXT value from DNS record [{}]", record);

@@ -0,0 +1,46 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog2.lookup.adapters.dnslookup;
+
+import com.github.joschi.jadconfig.Parameter;
+import com.github.joschi.jadconfig.util.Duration;
+import com.github.joschi.jadconfig.validators.PositiveDurationValidator;
+import com.github.joschi.jadconfig.validators.PositiveIntegerValidator;
+import org.graylog2.plugin.PluginConfigBean;
+
+public class DnsLookupAdapterConfiguration implements PluginConfigBean {
+    private static final String PREFIX = "dns_lookup_adapter_";
+    protected static final String RESOLVER_POOL_SIZE = PREFIX + "resolver_pool_size";
+    protected static final String RESOLVER_POOL_REFRESH_INTERVAL = PREFIX + "resolver_pool_refresh_interval";
+
+    protected static final int DEFAULT_POOL_SIZE = 10;
+    protected static final int DEFAULT_REFRESH_INTERVAL_SECONDS = 300;
+
+    @Parameter(value = RESOLVER_POOL_SIZE, validators = PositiveIntegerValidator.class)
+    private int poolSize = DEFAULT_POOL_SIZE;
+
+    @Parameter(value = RESOLVER_POOL_REFRESH_INTERVAL, validators = PositiveDurationValidator.class)
+    private Duration poolRefreshInterval = Duration.seconds(DEFAULT_REFRESH_INTERVAL_SECONDS);
+
+    public int getPoolSize() {
+        return poolSize;
+    }
+
+    public Duration getPoolRefreshInterval() {
+        return poolRefreshInterval;
+    }
+}

@@ -0,0 +1,82 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog2.lookup.adapters.dnslookup;
+
+import com.google.common.base.Splitter;
+import com.google.common.net.HostAndPort;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioDatagramChannel;
+import io.netty.resolver.dns.DnsNameResolver;
+import io.netty.resolver.dns.DnsNameResolverBuilder;
+import io.netty.resolver.dns.DnsServerAddressStreamProvider;
+import io.netty.resolver.dns.SequentialDnsServerAddressStreamProvider;
+import org.apache.commons.collections4.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+public class DnsNameResolverFactory {
+    private static final Logger LOG = LoggerFactory.getLogger(DnsNameResolverFactory.class);
+    private static final int DEFAULT_DNS_PORT = 53;
+
+    private final NioEventLoopGroup eventLoopGroup;
+    private String dnsServerIps;
+    private final long queryTimeout;
+
+    public DnsNameResolverFactory(NioEventLoopGroup eventLoopGroup, String dnsServerIps, long queryTimeout) {
+        this.eventLoopGroup = eventLoopGroup;
+        this.dnsServerIps = dnsServerIps;
+        this.queryTimeout = queryTimeout;
+    }
+
+    public DnsNameResolver create() {
+        final List<InetSocketAddress> iNetDnsServerIps = parseServerIpAddresses(dnsServerIps);
+        final DnsNameResolverBuilder dnsNameResolverBuilder = new DnsNameResolverBuilder(eventLoopGroup.next());
+        dnsNameResolverBuilder.channelType(NioDatagramChannel.class).queryTimeoutMillis(queryTimeout);
+
+        // Specify custom DNS servers if provided. If not, use those specified in local network adapter settings.
+        if (CollectionUtils.isNotEmpty(iNetDnsServerIps)) {
+            LOG.debug("Attempting to start DNS client with server IPs [{}] on port [{}].",
+                    dnsServerIps, DEFAULT_DNS_PORT);
+
+            final DnsServerAddressStreamProvider dnsServer = new SequentialDnsServerAddressStreamProvider(iNetDnsServerIps);
+            dnsNameResolverBuilder.nameServerProvider(dnsServer);
+        } else {
+            LOG.debug("Attempting to start DNS client with custom server IPs [{}] on port [{}].",
+                    dnsServerIps, DEFAULT_DNS_PORT);
+        }
+
+        return dnsNameResolverBuilder.build();
+    }
+
+    private List<InetSocketAddress> parseServerIpAddresses(String dnsServerIps) {
+
+        // Parse and prepare DNS server IP addresses for Netty.
+        return StreamSupport
+                // Split comma-separated sever IP:port combos.
+                .stream(Splitter.on(",").trimResults().omitEmptyStrings().split(dnsServerIps).spliterator(), false)
+                // Parse as HostAndPort objects (allows convenient handling of port provided after colon).
+                .map(hostAndPort -> HostAndPort.fromString(hostAndPort).withDefaultPort(DEFAULT_DNS_PORT))
+                // Convert HostAndPort > InetSocketAddress as required by Netty.
+                .map(hostAndPort -> new InetSocketAddress(hostAndPort.getHost(), hostAndPort.getPort()))
+                .collect(Collectors.toList());
+    }
+}

@@ -0,0 +1,242 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog2.lookup.adapters.dnslookup;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.resolver.dns.DnsNameResolver;
+import io.netty.util.concurrent.Future;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+/**
+ * Manages a pool of Netty {@link DnsNameResolverFactory} objects.
+ *
+ * <br>
+ * Since the source port for DNS resolution requests is fixed for the duration of the resolver lifecycle,
+ * the pooling capability of this class allows the source address to by varied for each request by choosing
+ * a random resolver from the pool for each request.
+ *
+ * <br>
+ * The resolvers in the pool are periodically refreshed to cycle in new source ports for subsequent requests.
+ *
+ * <br>
+ * The pool size and refresh interval are configurable globally for all DNS Lookup adapters with the following
+ * Graylog server configuration properties (the defaults are indicated below as well).
+ * <br>
+ * <pre>
+ * dns_lookup_adapter_resolver_pool_size = 10
+ * dns_lookup_adapter_resolver_pool_refresh_interval = 300s
+ * </pre>
+ *
+ * <br>
+ * Callers can use the {@link #takeLease()} method to acquire a lease for a resolver.
+ * The {@link ResolverLease#release()} method must be called to release a resolver lease after use.
+ * These operations are thread-safe.
+ */
+public class DnsResolverPool {
+    private static final Logger LOG = LoggerFactory.getLogger(DnsResolverPool.class);
+    private final long poolSize;
+    private final long poolRefreshSeconds;
+    private final ScheduledExecutorService executorService;
+    private final NioEventLoopGroup eventLoopGroup;
+    private final DnsNameResolverFactory resolverFactory;
+
+    // Pool is accessed by resolution requests and by refresh tasks.
+    private final List<ResolverLease> resolverPool;
+
+    protected DnsResolverPool(String dnsServerIps, long queryTimeout, long poolSize, long poolRefreshSeconds) {
+        this.poolSize = poolSize;
+        this.poolRefreshSeconds = poolRefreshSeconds;
+        this.executorService = Executors.newSingleThreadScheduledExecutor(
+                new ThreadFactoryBuilder().setNameFormat("dns-lookup-refresh-task-%d").build());
+
+        // A synchronized list is used to ensure thread safety for list mutations and accesses.
+        this.resolverPool = Collections.synchronizedList(new ArrayList<>());
+
+        // Use a single Netty EventLoopGroup (1 thread by default) for all resolvers in the pool.
+        // We are expecting DNS resolution requests for a specific pooler to be executed sequentially
+        // (one after the other) amongst the pool of resolvers. So, a single thread should be sufficient.
+        this.eventLoopGroup = new NioEventLoopGroup();
+        this.resolverFactory = new DnsNameResolverFactory(eventLoopGroup, dnsServerIps, queryTimeout);
+    }
+
+    protected void initialize() {
+        for (int i = 0; i < poolSize; i++) {
+            resolverPool.add(new ResolverLease(resolverFactory.create()));
+        }
+        executorService.scheduleAtFixedRate(new ResolverRefreshTask(), poolRefreshSeconds, poolRefreshSeconds, TimeUnit.SECONDS);
+    }
+
+    protected ResolverLease takeLease() {
+        if (resolverPool.size() == 0) {
+            throw new RuntimeException("Resolver pool is empty. Cannot return lease.");
+        }
+        final ResolverLease lease = resolverPool.get(randomResolverIndex());
+        lease.take();
+        return lease;
+    }
+
+    protected void returnLease(ResolverLease lease) {
+        lease.release();
+    }
+
+    public void stop() {
+        LOG.debug("Attempting to stop pool.");
+        executorService.shutdown();
+        if (resolverPool == null) {
+            LOG.error("Resolver pool has not been initialized.");
+            return;
+        }
+
+        synchronized (resolverPool) {
+            final Iterator<ResolverLease> iterator = resolverPool.iterator();
+            while (iterator.hasNext()) {
+                ResolverLease lease = iterator.next();
+                LOG.debug("Attempting to stop resolver [{}].", lease.getId());
+                if (lease.isLeased()) {
+                    LOG.warn("Attempting to stop a leased resolver...");
+                }
+                lease.take();
+                lease.getResolver().close();
+                iterator.remove();
+                LOG.debug("Successfully stopped resolver [{}].", lease.getId());
+            }
+        }
+
+        // Shutdown event loop (required by Netty).
+        final Future<?> shutdownFuture = eventLoopGroup.shutdownGracefully();
+        shutdownFuture.addListener(future -> LOG.debug("Finished shutting down pool."));
+        LOG.debug("Resolver pool shutdown complete.");
+    }
+
+    protected boolean isStopped() {
+        return (eventLoopGroup == null || eventLoopGroup.isShutdown()) && executorService.isShutdown();
+    }
+
+    /**
+     * Allows for a random resolver to be returned for each request.
+     */
+    protected int randomResolverIndex() {
+        // Use ThreadLocalRandom to ensure that different random numbers are returned when queried from different
+        // threads referencing the same instance.
+        return ThreadLocalRandom.current().nextInt(resolverPool.size());
+
+    }
+
+    private class ResolverRefreshTask implements Runnable {
+        @Override
+        public void run() {
+            LOG.debug("Starting resolver refresh.");
+            LOG.debug("Existing IDs: [{}]", resolverPool.stream().map(ResolverLease::getId).collect(Collectors.joining(", ")));
+            synchronized (resolverPool) {
+                final ListIterator<ResolverLease> iterator = resolverPool.listIterator();
+                while (iterator.hasNext()) {
+                    ResolverLease lease = iterator.next();
+                    if (!lease.getHasBeenLeased()) {
+                        LOG.debug("Resolver [{}] has not been leased yet. Skipping refresh.", lease.getId());
+                        continue;
+                    }
+
+                    if (!lease.isLeased()) {
+                        lease.getResolver().close();
+                        iterator.remove();
+                        iterator.add(new ResolverLease(resolverFactory.create()));
+                    } else {
+                        LOG.warn("Lease for resolver [{}] is in-use. Skipping refresh. This will be attempted again in [{}] seconds. " +
+                                        "If this happens frequently for high message rates, consider increasing the [dns_lookup_adapter_resolver_pool_size = 10] " +
+                                        "server configuration property to allow more DNS resolvers.",
+                                lease.getId(), poolRefreshSeconds);
+                    }
+                }
+            }
+            LOG.debug("Resolver IDs refreshed: [{}]", resolverPool.stream().map(ResolverLease::getId).collect(Collectors.joining(", ")));
+            LOG.debug("Finished resolver refresh.");
+        }
+    }
+
+    protected int poolSize() {
+        return resolverPool != null ? resolverPool.size() : 0;
+    }
+
+    protected static class ResolverLease {
+        private final String id;
+        private final DnsNameResolver resolver;
+        private AtomicInteger leaseCount;
+        private AtomicBoolean hasBeenLeased;
+
+        private ResolverLease(DnsNameResolver resolver) {
+            this.id = UUID.randomUUID().toString();
+            this.resolver = resolver;
+            this.leaseCount = new AtomicInteger(0);
+            this.hasBeenLeased = new AtomicBoolean();
+        }
+
+        private void take() {
+            this.leaseCount.incrementAndGet();
+            this.hasBeenLeased.set(true);
+        }
+
+        private void release() {
+            this.leaseCount.decrementAndGet();
+        }
+
+        protected String getId() {
+            return id;
+        }
+
+        private boolean isLeased() {
+            return leaseCount.get() > 0;
+        }
+
+        private boolean getHasBeenLeased() {
+            return hasBeenLeased.get();
+        }
+
+        protected DnsNameResolver getResolver() {
+            return resolver;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            final ResolverLease that = (ResolverLease) o;
+            return Objects.equals(id, that.id);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(id);
+        }
+    }
+}

@@ -0,0 +1,40 @@
+package org.graylog2.lookup.adapters.dnslookup;
+
+import org.graylog2.lookup.adapters.dnslookup.DnsResolverPool.ResolverLease;
+import org.junit.jupiter.api.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@RunWith(MockitoJUnitRunner.class)
+class DnsResolverPoolTest {
+
+    private static final int POOL_SIZE = 10;
+
+    @Test
+    public void testResolverIndex() {
+        final DnsResolverPool dnsResolverPool = new DnsResolverPool("", 100, POOL_SIZE, 300);
+        dnsResolverPool.initialize();
+        assertEquals(POOL_SIZE, dnsResolverPool.poolSize());
+        final int randomResolverIndex = dnsResolverPool.randomResolverIndex();
+        assertTrue(randomResolverIndex >= 0);
+        assertTrue(randomResolverIndex < POOL_SIZE);
+        dnsResolverPool.stop();
+    }
+
+    @Test
+    public void verifyRefresh() throws InterruptedException {
+        final DnsResolverPool dnsResolverPool = new DnsResolverPool("", 100, 1, 1);
+        dnsResolverPool.initialize();
+        final ResolverLease nextLease = dnsResolverPool.takeLease();
+        final String initialLeaseId = nextLease.getId();
+        dnsResolverPool.returnLease(nextLease);
+        Thread.sleep(1500);
+        final ResolverLease finalLease = dnsResolverPool.takeLease();
+        assertNotEquals(initialLeaseId, finalLease.getId());
+        dnsResolverPool.stop();
+    }
+}