/*
 * 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.facebook.presto.hive;

import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.net.HostAndPort;
import com.google.common.primitives.Ints;
import io.airlift.units.Duration;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.metrics.ContextFactory;
import org.apache.hadoop.net.DNSToSwitchMapping;
import org.apache.hadoop.net.SocksSocketFactory;

import javax.inject.Inject;
import javax.net.SocketFactory;

import java.io.IOException;
import java.util.List;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;

public class HdfsConfiguration
{
    private final HostAndPort socksProxy;
    private final Duration dfsTimeout;
    private final Duration dfsConnectTimeout;
    private final int dfsConnectMaxRetries;
    private final String domainSocketPath;

    @SuppressWarnings("ThreadLocalNotStaticFinal")
    private final ThreadLocal<Configuration> hadoopConfiguration = new ThreadLocal<Configuration>()
    {
        @Override
        protected Configuration initialValue()
        {
            return createConfiguration();
        }
    };

    @Inject
    public HdfsConfiguration(HiveClientConfig hiveClientConfig)
    {
        checkNotNull(hiveClientConfig, "hiveClientConfig is null");
        checkArgument(hiveClientConfig.getDfsTimeout().toMillis() >= 1, "dfsTimeout must be at least 1 ms");

        this.socksProxy = hiveClientConfig.getMetastoreSocksProxy();
        this.dfsTimeout = hiveClientConfig.getDfsTimeout();
        this.dfsConnectTimeout = hiveClientConfig.getDfsConnectTimeout();
        this.dfsConnectMaxRetries = hiveClientConfig.getDfsConnectMaxRetries();
        this.domainSocketPath = hiveClientConfig.getDomainSocketPath();
    }

    @SuppressWarnings("UnusedParameters")
    public Configuration getConfiguration(String host)
    {
        // subclasses can provide per-host configuration
        return hadoopConfiguration.get();
    }

    protected Configuration createConfiguration()
    {
        Configuration config = new Configuration();

        // this is to prevent dfs client from doing reverse DNS lookups to determine whether nodes are rack local
        config.setClass("topology.node.switch.mapping.impl", NoOpDNSToSwitchMapping.class, DNSToSwitchMapping.class);

        if (socksProxy != null) {
            config.setClass("hadoop.rpc.socket.factory.class.default", SocksSocketFactory.class, SocketFactory.class);
            config.set("hadoop.socks.server", socksProxy.toString());
        }

        config.setBoolean("dfs.client.read.shortcircuit", true);
        if (domainSocketPath != null) {
            config.setStrings("dfs.domain.socket.path", domainSocketPath);
        }

        config.setBoolean("dfs.read.shortcircuit", true);
        config.setBoolean("dfs.read.shortcircuit.fallbackwhenfail", true);
        config.setInt("dfs.socket.timeout", Ints.checkedCast(dfsTimeout.toMillis()));
        config.setInt("ipc.ping.interval", Ints.checkedCast(dfsTimeout.toMillis()));
        config.setInt("ipc.client.connect.timeout", Ints.checkedCast(dfsConnectTimeout.toMillis()));
        config.setInt("ipc.client.connect.max.retries", dfsConnectMaxRetries);

        // Enable JMX export of stats for DFSClient
        try {
            ContextFactory factory = ContextFactory.getFactory();
            factory.setAttribute("hdfsclient.class", "org.apache.hadoop.metrics.jmx.JMXContext");
            factory.setAttribute("hdfsclient.period", "5"); // Counter value publish interval (seconds)
        }
        catch (IOException e) {
            throw Throwables.propagate(e);
        }

        return config;
    }

    public static class NoOpDNSToSwitchMapping
            implements DNSToSwitchMapping
    {
        @Override
        public List<String> resolve(List<String> names)
        {
            // dfs client expects an empty list as an indication that the host->switch mapping for the given names are not known
            return ImmutableList.of();
        }

        @Override
        public void reloadCachedMappings()
        {
            // no-op
        }
    }
}

@@ -594,50 +594,8 @@
 e;%0A%0A
-import com.google.common.base.Throwables;%0A
 impo
@@ -638,16 +638,16 @@
 leList;%0A
+
 import c
@@ -805,57 +805,8 @@
 on;%0A
-import org.apache.hadoop.metrics.ContextFactory;%0A
 impo
@@ -965,36 +965,8 @@
 y;%0A%0A
-import java.io.IOException;%0A
 impo
@@ -3259,141 +3259,8 @@
  %7D%0A%0A
-        config.setBoolean(%22dfs.read.shortcircuit%22, true);%0A        config.setBoolean(%22dfs.read.shortcircuit.fallbackwhenfail%22, true);%0A
     
@@ -3619,452 +3619,236 @@
     
-// Enable JMX export of stats for DFSClient%0A        try %7B%0A            ContextFactory factory = ContextFactory.getFactory();%0A            factory.setAttribute(%22hdfsclient.class%22, %22org.apache.hadoop.metrics.jmx.JMXContext
+updateConfiguration(config);%0A%0A        return config;%0A    %7D%0A%0A    @SuppressWarnings(%22UnusedParameters
 %22)
-;
 %0A    
-        factory.setAttribute(%22hdfsclient.period%22, %225%22); // Counter value publish interval (seconds)%0A        %7D%0A        catch (IOException e) %7B%0A            throw Throwables.propagate(e);%0A        %7D%0A%0A        return config;
+protected void updateConfiguration(Configuration config)%0A    %7B%0A        // allow subclasses to modify configuration objects
 %0A   

Remove unused configuration