/*
 * Copyright 2010-2011 Ning, Inc.
 *
 * Ning licenses this file to you 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.ning.metrics.collector.hadoop.writer;

import com.google.inject.Inject;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;

public class FileSystemAccess
{
    private final Configuration hdfsConfig;
    private final Class<? extends FileSystem> fsClass;
    private FileSystem fs = null;
    private final Object connectionLock = new Object();

    private static final long MIN_WAIT_TIME = 1000; // 1 second
    private static final long MAX_WAIT_TIME = 600000; // 10 minutes

    @Inject
    public FileSystemAccess(final Configuration hdfsConfig)
    {
        this(hdfsConfig, FileSystem.class);
    }

    public FileSystemAccess(final Configuration hdfsConfig, final Class<? extends FileSystem> fsClass)
    {
        this.hdfsConfig = hdfsConfig;
        this.fsClass = fsClass;
    }

    public FileSystem get()
    {
        try {
            return getFileSystemSafe();
        }
        catch (IOException e) {
            synchronized (connectionLock) {
                long waitTime = MIN_WAIT_TIME;
                while (true) {
                    try {
                        // try to set up the FileSystem
                        setFileSystem();
                        return getFileSystemSafe();
                    }
                    catch (IOException e1) {
                        // exp backoff
                        try {
                            Thread.sleep(waitTime);
                        }
                        catch (InterruptedException e2) {
                            e2.printStackTrace();
                        }

                        if (waitTime <= MAX_WAIT_TIME / 2) {
                            waitTime *= 2;
                        }
                        else {
                            waitTime = MAX_WAIT_TIME;
                        }
                    }
                }
            }
        }
    }

    // throws an IOException if the current FileSystem isn't working
    private FileSystem getFileSystemSafe() throws IOException
    {
        try {
            fs.getFileStatus(new Path("/"));
            return fs;
        }
        catch (NullPointerException e) {
            throw new IOException("file system not initialized");
        }
    }

    private void setFileSystem() throws IOException
    {
        try {
            final Method getMethod = fsClass.getMethod("get", Configuration.class);
            fs = (FileSystem) getMethod.invoke(null, hdfsConfig);
        }
        catch (NoSuchMethodException e) {
            throw new IOException(String.format("Class %s doesn't have a get method", fsClass), e);
        }
        catch (InvocationTargetException e) {
            throw new IOException(String.format("Got exception while accessing get method for class %s ", fsClass), e);
        }
        catch (IllegalAccessException e) {
            throw new IOException(String.format("Got exception while accessing get method for class %s ", fsClass), e);
        }
        catch (RuntimeException e) {
            throw new IOException(String.format("Got exception while accessing get method for class %s ", fsClass), e);
        }
    }
}

@@ -818,16 +818,48 @@
 fs.Path;
+%0Aimport org.apache.log4j.Logger;
 %0A%0Aimport
@@ -998,16 +998,97 @@
 ccess%0A%7B%0A
+    private static final Logger log = Logger.getLogger(FileSystemAccess.class);%0A%0A
     priv
@@ -1839,32 +1839,144 @@
 OException e) %7B%0A
+            log.warn(String.format(%22Got IOException trying to connect to HDFS: %25s%22, e.getLocalizedMessage()));%0A%0A
             sync

add log message on connection exceptions