// Path: com.ibm.streamsx.hdfs/impl/java/src/com/ibm/streamsx/hdfs/client/IHdfsClient.java
// public interface IHdfsClient {
// 
// 	public FileSystem connect(String fileSystemUri, String hdfsUser, String configPath) throws Exception;
// 
// 	public InputStream getInputStream(String filePath) throws IOException;
// 
// 	public OutputStream getOutputStream(String filePath, boolean append) throws IOException;
// 
// 	public FileStatus[] scanDirectory(String dirPath, String filter) throws IOException;
// 
// 	public boolean exists(String filePath) throws IOException;
// 
// 	public boolean isDirectory(String filePath) throws IOException;
// 
// 	public long getFileSize(String filename) throws IOException;
// 
// 	public boolean rename(String src, String dst) throws IOException;
// 
// 	public boolean delete(String filePath, boolean recursive) throws IOException;
// 
// 	public void disconnect() throws Exception;
// 
// 	public void setConnectionProperty(String name, String value);
// 
// 	public String getConnectionProperty(String name);
// 
// 	public Map<String, String> getConnectionProperties();
// }


import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.logging.Logger;
import org.apache.hadoop.fs.FSDataInputStream;
import com.ibm.streams.operator.OperatorContext;
import com.ibm.streams.operator.OperatorContext.ContextCheck;
import com.ibm.streams.operator.OutputTuple;
import com.ibm.streams.operator.StreamSchema;
import com.ibm.streams.operator.StreamingData.Punctuation;
import com.ibm.streams.operator.StreamingInput;
import com.ibm.streams.operator.StreamingOutput;
import com.ibm.streams.operator.Tuple;
import com.ibm.streams.operator.Type.MetaType;
import com.ibm.streams.operator.compile.OperatorContextChecker;
import com.ibm.streams.operator.logging.LogLevel;
import com.ibm.streams.operator.logging.LoggerNames;
import com.ibm.streams.operator.logging.TraceLevel;
import com.ibm.streams.operator.metrics.Metric;
import com.ibm.streams.operator.model.Icons;
import com.ibm.streams.operator.model.InputPortSet;
import com.ibm.streams.operator.model.InputPorts;
import com.ibm.streams.operator.model.OutputPortSet;
import com.ibm.streams.operator.model.OutputPorts;
import com.ibm.streams.operator.model.Parameter;
import com.ibm.streams.operator.model.PrimitiveOperator;
import com.ibm.streams.operator.model.SharedLoader;
import com.ibm.streams.operator.state.Checkpoint;
import com.ibm.streams.operator.state.ConsistentRegionContext;
import com.ibm.streams.operator.state.StateHandler;
import com.ibm.streams.operator.types.ValueFactory;
import com.ibm.streams.operator.model.InputPortSet.WindowMode;
import com.ibm.streams.operator.model.InputPortSet.WindowPunctuationInputMode;
import com.ibm.streams.operator.model.OutputPortSet.WindowPunctuationOutputMode;
import com.ibm.streamsx.hdfs.client.IHdfsClient;
			if (fileUri.getScheme() != null) {
				// must have the same scheme
				if (!hdfsUri.getScheme().equals(fileUri.getScheme())) {
					checker.setInvalidContext(Messages.getString("HDFS_SOURCE_INVALID_SCHEMA", fileUri.getScheme(),
							hdfsUri.getScheme()), null);
					return;
				}

				// must have the same authority
				if ((hdfsUri.getAuthority() == null && fileUri.getAuthority() != null) || (hdfsUri
						.getAuthority() != null && fileUri.getAuthority() == null) || (hdfsUri.getAuthority() != null
								&& fileUri.getAuthority() != null && !hdfsUri.getAuthority().equals(fileUri
										.getAuthority()))) {
					checker.setInvalidContext(Messages.getString("HDFS_SOURCE_INVALID_HOST", fileUri.getAuthority(),
							hdfsUri.getAuthority()), null);
					return;
				}
			}
		}
	}

	private void initMetrics(OperatorContext context) {
		nOpenedFiles = context.getMetrics().createCustomMetric(FILES_OPENED_METRIC, "Number of opened files ",
				Metric.Kind.COUNTER);
	}

	private void processFile(String filename) throws Exception {
		if (LOGGER.isLoggable(LogLevel.INFO)) {
			LOGGER.log(LogLevel.INFO, Messages.getString("HDFS_SOURCE_PROCESS_FILE", filename));
		}
// Path: com.ibm.streamsx.hdfs/impl/java/src/com/ibm/streamsx/hdfs/client/IHdfsClient.java
// public interface IHdfsClient {
// 
// 	public FileSystem connect(String fileSystemUri, String hdfsUser, String configPath) throws Exception;
// 
// 	public InputStream getInputStream(String filePath) throws IOException;
// 
// 	public OutputStream getOutputStream(String filePath, boolean append) throws IOException;
// 
// 	public FileStatus[] scanDirectory(String dirPath, String filter) throws IOException;
// 
// 	public boolean exists(String filePath) throws IOException;
// 
// 	public boolean isDirectory(String filePath) throws IOException;
// 
// 	public long getFileSize(String filename) throws IOException;
// 
// 	public boolean rename(String src, String dst) throws IOException;
// 
// 	public boolean delete(String filePath, boolean recursive) throws IOException;
// 
// 	public void disconnect() throws Exception;
// 
// 	public void setConnectionProperty(String name, String value);
// 
// 	public String getConnectionProperty(String name);
// 
// 	public Map<String, String> getConnectionProperties();
// }


// Path: com.ibm.streamsx.hdfs/impl/java/src/com/ibm/streamsx/hdfs/HDFS2FileSource.java
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.logging.Logger;
import org.apache.hadoop.fs.FSDataInputStream;
import com.ibm.streams.operator.OperatorContext;
import com.ibm.streams.operator.OperatorContext.ContextCheck;
import com.ibm.streams.operator.OutputTuple;
import com.ibm.streams.operator.StreamSchema;
import com.ibm.streams.operator.StreamingData.Punctuation;
import com.ibm.streams.operator.StreamingInput;
import com.ibm.streams.operator.StreamingOutput;
import com.ibm.streams.operator.Tuple;
import com.ibm.streams.operator.Type.MetaType;
import com.ibm.streams.operator.compile.OperatorContextChecker;
import com.ibm.streams.operator.logging.LogLevel;
import com.ibm.streams.operator.logging.LoggerNames;
import com.ibm.streams.operator.logging.TraceLevel;
import com.ibm.streams.operator.metrics.Metric;
import com.ibm.streams.operator.model.Icons;
import com.ibm.streams.operator.model.InputPortSet;
import com.ibm.streams.operator.model.InputPorts;
import com.ibm.streams.operator.model.OutputPortSet;
import com.ibm.streams.operator.model.OutputPorts;
import com.ibm.streams.operator.model.Parameter;
import com.ibm.streams.operator.model.PrimitiveOperator;
import com.ibm.streams.operator.model.SharedLoader;
import com.ibm.streams.operator.state.Checkpoint;
import com.ibm.streams.operator.state.ConsistentRegionContext;
import com.ibm.streams.operator.state.StateHandler;
import com.ibm.streams.operator.types.ValueFactory;
import com.ibm.streams.operator.model.InputPortSet.WindowMode;
import com.ibm.streams.operator.model.InputPortSet.WindowPunctuationInputMode;
import com.ibm.streams.operator.model.OutputPortSet.WindowPunctuationOutputMode;
import com.ibm.streamsx.hdfs.client.IHdfsClient;

			if (fileUri.getScheme() != null) {
				// must have the same scheme
				if (!hdfsUri.getScheme().equals(fileUri.getScheme())) {
					checker.setInvalidContext(Messages.getString("HDFS_SOURCE_INVALID_SCHEMA", fileUri.getScheme(),
							hdfsUri.getScheme()), null);
					return;
				}

				// must have the same authority
				if ((hdfsUri.getAuthority() == null && fileUri.getAuthority() != null) || (hdfsUri
						.getAuthority() != null && fileUri.getAuthority() == null) || (hdfsUri.getAuthority() != null
								&& fileUri.getAuthority() != null && !hdfsUri.getAuthority().equals(fileUri
										.getAuthority()))) {
					checker.setInvalidContext(Messages.getString("HDFS_SOURCE_INVALID_HOST", fileUri.getAuthority(),
							hdfsUri.getAuthority()), null);
					return;
				}
			}
		}
	}

	private void initMetrics(OperatorContext context) {
		nOpenedFiles = context.getMetrics().createCustomMetric(FILES_OPENED_METRIC, "Number of opened files ",
				Metric.Kind.COUNTER);
	}

	private void processFile(String filename) throws Exception {
		if (LOGGER.isLoggable(LogLevel.INFO)) {
			LOGGER.log(LogLevel.INFO, Messages.getString("HDFS_SOURCE_PROCESS_FILE", filename));
		}
		IHdfsClient hdfsClient = getHdfsClient();