/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF 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 accumulo;

import java.io.File;

import com.google.common.base.Preconditions;

/**
 * 
 */
public class AccumuloCsvDirectoryInput implements AccumuloCsvInput {

  protected final File directory;
  
  public AccumuloCsvDirectoryInput(File dir) {
    Preconditions.checkArgument(dir.exists() && dir.isDirectory());
    
    directory = dir;
  }

  @Override
  public Iterable<File> getInputFiles() {
    throw new UnsupportedOperationException();
  }
}

@@ -834,16 +834,122 @@
 io.File;
+%0Aimport java.util.HashSet;%0Aimport java.util.Set;%0A%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;
 %0A%0Aimport
@@ -1069,17 +1069,110 @@
 Input %7B%0A
+  private static final Logger log = LoggerFactory.getLogger(AccumuloCsvDirectoryInput.class);
 %0A
-
   protec
@@ -1408,50 +1408,555 @@
     
-throw new UnsupportedOperationException();
+Set%3CFile%3E files = new HashSet%3CFile%3E();%0A    getFiles(files, directory);%0A    return files;%0A  %7D%0A  %0A  protected void getFiles(Set%3CFile%3E files, File f) %7B%0A    for (String child : f.list()) %7B%0A      File childFile = new File(f, child);%0A      if (files.contains(childFile)) %7B%0A        // continue%0A      %7D else if (childFile.isDirectory()) %7B%0A        getFiles(files, childFile);%0A      %7D else if (childFile.isFile()) %7B%0A        files.add(childFile);%0A      %7D else %7B%0A        log.warn(%22Ignoring %7B%7D because it's not a file nor directory.%22, childFile);%0A      %7D%0A    %7D
 %0A  %7D

Implement a simple recursive directory traversal (no cycle detection)