@Test(expected = IOException.class) public void convert_nopath() throws Exception { Map<String, String> attributes = new HashMap<>(); DirectDataSourceProfile profile = new DirectDataSourceProfile( "testing", HadoopDataSource.class, "context", attributes); Configuration conf = new Configuration(); HadoopDataSourceProfile.convert(profile, conf); }
public static HadoopDataSourceProfile convert( DirectDataSourceProfile profile, Configuration conf) throws IOException { if (profile == null) { throw new IllegalArgumentException("profile must not be null"); } if (conf == null) { throw new IllegalArgumentException("conf must not be null"); } Map<String, String> attributes = new HashMap<>(profile.getAttributes()); Path fsPath = takeFsPath(profile, attributes, conf); if (fsPath == null) { throw new IOException(MessageFormat.format( "The directio configuration \"{0} ({1})\" does not have \"{2}\"", profile.getId(), profile.getPath().isEmpty() ? ROOT_REPRESENTATION : profile.getPath(), fqn(profile, KEY_PATH))); } Path tempPath = takeTempPath(profile, attributes, conf, fsPath); FileSystem fileSystem = fsPath.getFileSystem(conf); FileSystem tempFs = tempPath.getFileSystem(conf); if (getFsIdentity(fileSystem).equals(getFsIdentity(tempFs)) == false) { throw new IOException(MessageFormat.format( "The directio target and temporary path must be on same file system ({0}={1} <=> {2}={3})", fqn(profile, KEY_PATH), fsPath, fqn(profile, KEY_TEMP), tempPath)); } fsPath = fileSystem.makeQualified(fsPath); tempPath = fileSystem.makeQualified(tempPath); HadoopDataSourceProfile result = new HadoopDataSourceProfile( conf, profile.getId(), profile.getPath(), fsPath, tempPath); long minFragment = takeMinFragment(profile, attributes, conf); result.setMinimumFragmentSize(minFragment); long prefFragment = takePrefFragment(profile, attributes, conf); result.setPreferredFragmentSize(prefFragment); result.setOutputStaging(takeBoolean(profile, attributes, KEY_OUTPUT_STAGING, DEFAULT_OUTPUT_STAGING)); result.setOutputStreaming(takeBoolean(profile, attributes, KEY_OUTPUT_STREAMING, DEFAULT_OUTPUT_STREAMING)); result.setSplitBlocks(takeBoolean(profile, attributes, KEY_SPLIT_BLOCKS, DEFAULT_SPLIT_BLOCKS)); result.setCombineBlocks(takeBoolean(profile, attributes, KEY_COMBINE_BLOCKS, DEFAULT_COMBINE_BLOCKS)); result.setKeepAliveInterval( takePositive(profile, attributes, KEY_KEEPALIVE_INTERVAL, DEFAULT_KEEPALIVE_INTERVAL)); result.setRollforwardThreads( takePositive(profile, attributes, KEY_ROLLFORWARD_THREADS, DEFAULT_ROLLFORWARD_THREADS)); if (attributes.isEmpty() == false) { throw new IOException(MessageFormat.format( "Unknown attributes in \"{0}\": {1}", profile.getId(), new TreeSet<>(attributes.keySet()))); } return result; }
HadoopDataSourceProfile { public static HadoopDataSourceProfile convert( DirectDataSourceProfile profile, Configuration conf) throws IOException { if (profile == null) { throw new IllegalArgumentException("profile must not be null"); } if (conf == null) { throw new IllegalArgumentException("conf must not be null"); } Map<String, String> attributes = new HashMap<>(profile.getAttributes()); Path fsPath = takeFsPath(profile, attributes, conf); if (fsPath == null) { throw new IOException(MessageFormat.format( "The directio configuration \"{0} ({1})\" does not have \"{2}\"", profile.getId(), profile.getPath().isEmpty() ? ROOT_REPRESENTATION : profile.getPath(), fqn(profile, KEY_PATH))); } Path tempPath = takeTempPath(profile, attributes, conf, fsPath); FileSystem fileSystem = fsPath.getFileSystem(conf); FileSystem tempFs = tempPath.getFileSystem(conf); if (getFsIdentity(fileSystem).equals(getFsIdentity(tempFs)) == false) { throw new IOException(MessageFormat.format( "The directio target and temporary path must be on same file system ({0}={1} <=> {2}={3})", fqn(profile, KEY_PATH), fsPath, fqn(profile, KEY_TEMP), tempPath)); } fsPath = fileSystem.makeQualified(fsPath); tempPath = fileSystem.makeQualified(tempPath); HadoopDataSourceProfile result = new HadoopDataSourceProfile( conf, profile.getId(), profile.getPath(), fsPath, tempPath); long minFragment = takeMinFragment(profile, attributes, conf); result.setMinimumFragmentSize(minFragment); long prefFragment = takePrefFragment(profile, attributes, conf); result.setPreferredFragmentSize(prefFragment); result.setOutputStaging(takeBoolean(profile, attributes, KEY_OUTPUT_STAGING, DEFAULT_OUTPUT_STAGING)); result.setOutputStreaming(takeBoolean(profile, attributes, KEY_OUTPUT_STREAMING, DEFAULT_OUTPUT_STREAMING)); result.setSplitBlocks(takeBoolean(profile, attributes, KEY_SPLIT_BLOCKS, DEFAULT_SPLIT_BLOCKS)); result.setCombineBlocks(takeBoolean(profile, attributes, KEY_COMBINE_BLOCKS, DEFAULT_COMBINE_BLOCKS)); result.setKeepAliveInterval( takePositive(profile, attributes, KEY_KEEPALIVE_INTERVAL, DEFAULT_KEEPALIVE_INTERVAL)); result.setRollforwardThreads( takePositive(profile, attributes, KEY_ROLLFORWARD_THREADS, DEFAULT_ROLLFORWARD_THREADS)); if (attributes.isEmpty() == false) { throw new IOException(MessageFormat.format( "Unknown attributes in \"{0}\": {1}", profile.getId(), new TreeSet<>(attributes.keySet()))); } return result; } }
HadoopDataSourceProfile { public static HadoopDataSourceProfile convert( DirectDataSourceProfile profile, Configuration conf) throws IOException { if (profile == null) { throw new IllegalArgumentException("profile must not be null"); } if (conf == null) { throw new IllegalArgumentException("conf must not be null"); } Map<String, String> attributes = new HashMap<>(profile.getAttributes()); Path fsPath = takeFsPath(profile, attributes, conf); if (fsPath == null) { throw new IOException(MessageFormat.format( "The directio configuration \"{0} ({1})\" does not have \"{2}\"", profile.getId(), profile.getPath().isEmpty() ? ROOT_REPRESENTATION : profile.getPath(), fqn(profile, KEY_PATH))); } Path tempPath = takeTempPath(profile, attributes, conf, fsPath); FileSystem fileSystem = fsPath.getFileSystem(conf); FileSystem tempFs = tempPath.getFileSystem(conf); if (getFsIdentity(fileSystem).equals(getFsIdentity(tempFs)) == false) { throw new IOException(MessageFormat.format( "The directio target and temporary path must be on same file system ({0}={1} <=> {2}={3})", fqn(profile, KEY_PATH), fsPath, fqn(profile, KEY_TEMP), tempPath)); } fsPath = fileSystem.makeQualified(fsPath); tempPath = fileSystem.makeQualified(tempPath); HadoopDataSourceProfile result = new HadoopDataSourceProfile( conf, profile.getId(), profile.getPath(), fsPath, tempPath); long minFragment = takeMinFragment(profile, attributes, conf); result.setMinimumFragmentSize(minFragment); long prefFragment = takePrefFragment(profile, attributes, conf); result.setPreferredFragmentSize(prefFragment); result.setOutputStaging(takeBoolean(profile, attributes, KEY_OUTPUT_STAGING, DEFAULT_OUTPUT_STAGING)); result.setOutputStreaming(takeBoolean(profile, attributes, KEY_OUTPUT_STREAMING, DEFAULT_OUTPUT_STREAMING)); result.setSplitBlocks(takeBoolean(profile, attributes, KEY_SPLIT_BLOCKS, DEFAULT_SPLIT_BLOCKS)); result.setCombineBlocks(takeBoolean(profile, attributes, KEY_COMBINE_BLOCKS, DEFAULT_COMBINE_BLOCKS)); result.setKeepAliveInterval( takePositive(profile, attributes, KEY_KEEPALIVE_INTERVAL, DEFAULT_KEEPALIVE_INTERVAL)); result.setRollforwardThreads( takePositive(profile, attributes, KEY_ROLLFORWARD_THREADS, DEFAULT_ROLLFORWARD_THREADS)); if (attributes.isEmpty() == false) { throw new IOException(MessageFormat.format( "Unknown attributes in \"{0}\": {1}", profile.getId(), new TreeSet<>(attributes.keySet()))); } return result; } HadoopDataSourceProfile(
            Configuration conf,
            String id,
            String contextPath,
            Path fileSystemPath,
            Path temporaryPath); }
HadoopDataSourceProfile { public static HadoopDataSourceProfile convert( DirectDataSourceProfile profile, Configuration conf) throws IOException { if (profile == null) { throw new IllegalArgumentException("profile must not be null"); } if (conf == null) { throw new IllegalArgumentException("conf must not be null"); } Map<String, String> attributes = new HashMap<>(profile.getAttributes()); Path fsPath = takeFsPath(profile, attributes, conf); if (fsPath == null) { throw new IOException(MessageFormat.format( "The directio configuration \"{0} ({1})\" does not have \"{2}\"", profile.getId(), profile.getPath().isEmpty() ? ROOT_REPRESENTATION : profile.getPath(), fqn(profile, KEY_PATH))); } Path tempPath = takeTempPath(profile, attributes, conf, fsPath); FileSystem fileSystem = fsPath.getFileSystem(conf); FileSystem tempFs = tempPath.getFileSystem(conf); if (getFsIdentity(fileSystem).equals(getFsIdentity(tempFs)) == false) { throw new IOException(MessageFormat.format( "The directio target and temporary path must be on same file system ({0}={1} <=> {2}={3})", fqn(profile, KEY_PATH), fsPath, fqn(profile, KEY_TEMP), tempPath)); } fsPath = fileSystem.makeQualified(fsPath); tempPath = fileSystem.makeQualified(tempPath); HadoopDataSourceProfile result = new HadoopDataSourceProfile( conf, profile.getId(), profile.getPath(), fsPath, tempPath); long minFragment = takeMinFragment(profile, attributes, conf); result.setMinimumFragmentSize(minFragment); long prefFragment = takePrefFragment(profile, attributes, conf); result.setPreferredFragmentSize(prefFragment); result.setOutputStaging(takeBoolean(profile, attributes, KEY_OUTPUT_STAGING, DEFAULT_OUTPUT_STAGING)); result.setOutputStreaming(takeBoolean(profile, attributes, KEY_OUTPUT_STREAMING, DEFAULT_OUTPUT_STREAMING)); result.setSplitBlocks(takeBoolean(profile, attributes, KEY_SPLIT_BLOCKS, DEFAULT_SPLIT_BLOCKS)); result.setCombineBlocks(takeBoolean(profile, attributes, KEY_COMBINE_BLOCKS, DEFAULT_COMBINE_BLOCKS)); result.setKeepAliveInterval( takePositive(profile, attributes, KEY_KEEPALIVE_INTERVAL, DEFAULT_KEEPALIVE_INTERVAL)); result.setRollforwardThreads( takePositive(profile, attributes, KEY_ROLLFORWARD_THREADS, DEFAULT_ROLLFORWARD_THREADS)); if (attributes.isEmpty() == false) { throw new IOException(MessageFormat.format( "Unknown attributes in \"{0}\": {1}", profile.getId(), new TreeSet<>(attributes.keySet()))); } return result; } HadoopDataSourceProfile(
            Configuration conf,
            String id,
            String contextPath,
            Path fileSystemPath,
            Path temporaryPath); String getId(); String getContextPath(); Path getFileSystemPath(); Path getTemporaryFileSystemPath(); FileSystem getFileSystem(); LocalFileSystem getLocalFileSystem(); long getMinimumFragmentSize(FragmentableDataFormat<?> format); long getMinimumFragmentSize(); void setMinimumFragmentSize(long size); long getPreferredFragmentSize(FragmentableDataFormat<?> format); long getPreferredFragmentSize(); void setPreferredFragmentSize(long size); boolean isSplitBlocks(); void setSplitBlocks(boolean split); boolean isCombineBlocks(); void setCombineBlocks(boolean combine); boolean isOutputStaging(); void setOutputStaging(boolean required); boolean isOutputStreaming(); void setOutputStreaming(boolean required); long getKeepAliveInterval(); void setKeepAliveInterval(long interval); int getRollforwardThreads(); void setRollforwardThreads(int threads); @Override String toString(); static HadoopDataSourceProfile convert(
            DirectDataSourceProfile profile,
            Configuration conf); }
HadoopDataSourceProfile { public static HadoopDataSourceProfile convert( DirectDataSourceProfile profile, Configuration conf) throws IOException { if (profile == null) { throw new IllegalArgumentException("profile must not be null"); } if (conf == null) { throw new IllegalArgumentException("conf must not be null"); } Map<String, String> attributes = new HashMap<>(profile.getAttributes()); Path fsPath = takeFsPath(profile, attributes, conf); if (fsPath == null) { throw new IOException(MessageFormat.format( "The directio configuration \"{0} ({1})\" does not have \"{2}\"", profile.getId(), profile.getPath().isEmpty() ? ROOT_REPRESENTATION : profile.getPath(), fqn(profile, KEY_PATH))); } Path tempPath = takeTempPath(profile, attributes, conf, fsPath); FileSystem fileSystem = fsPath.getFileSystem(conf); FileSystem tempFs = tempPath.getFileSystem(conf); if (getFsIdentity(fileSystem).equals(getFsIdentity(tempFs)) == false) { throw new IOException(MessageFormat.format( "The directio target and temporary path must be on same file system ({0}={1} <=> {2}={3})", fqn(profile, KEY_PATH), fsPath, fqn(profile, KEY_TEMP), tempPath)); } fsPath = fileSystem.makeQualified(fsPath); tempPath = fileSystem.makeQualified(tempPath); HadoopDataSourceProfile result = new HadoopDataSourceProfile( conf, profile.getId(), profile.getPath(), fsPath, tempPath); long minFragment = takeMinFragment(profile, attributes, conf); result.setMinimumFragmentSize(minFragment); long prefFragment = takePrefFragment(profile, attributes, conf); result.setPreferredFragmentSize(prefFragment); result.setOutputStaging(takeBoolean(profile, attributes, KEY_OUTPUT_STAGING, DEFAULT_OUTPUT_STAGING)); result.setOutputStreaming(takeBoolean(profile, attributes, KEY_OUTPUT_STREAMING, DEFAULT_OUTPUT_STREAMING)); result.setSplitBlocks(takeBoolean(profile, attributes, KEY_SPLIT_BLOCKS, DEFAULT_SPLIT_BLOCKS)); result.setCombineBlocks(takeBoolean(profile, attributes, KEY_COMBINE_BLOCKS, DEFAULT_COMBINE_BLOCKS)); result.setKeepAliveInterval( takePositive(profile, attributes, KEY_KEEPALIVE_INTERVAL, DEFAULT_KEEPALIVE_INTERVAL)); result.setRollforwardThreads( takePositive(profile, attributes, KEY_ROLLFORWARD_THREADS, DEFAULT_ROLLFORWARD_THREADS)); if (attributes.isEmpty() == false) { throw new IOException(MessageFormat.format( "Unknown attributes in \"{0}\": {1}", profile.getId(), new TreeSet<>(attributes.keySet()))); } return result; } HadoopDataSourceProfile(
            Configuration conf,
            String id,
            String contextPath,
            Path fileSystemPath,
            Path temporaryPath); String getId(); String getContextPath(); Path getFileSystemPath(); Path getTemporaryFileSystemPath(); FileSystem getFileSystem(); LocalFileSystem getLocalFileSystem(); long getMinimumFragmentSize(FragmentableDataFormat<?> format); long getMinimumFragmentSize(); void setMinimumFragmentSize(long size); long getPreferredFragmentSize(FragmentableDataFormat<?> format); long getPreferredFragmentSize(); void setPreferredFragmentSize(long size); boolean isSplitBlocks(); void setSplitBlocks(boolean split); boolean isCombineBlocks(); void setCombineBlocks(boolean combine); boolean isOutputStaging(); void setOutputStaging(boolean required); boolean isOutputStreaming(); void setOutputStreaming(boolean required); long getKeepAliveInterval(); void setKeepAliveInterval(long interval); int getRollforwardThreads(); void setRollforwardThreads(int threads); @Override String toString(); static HadoopDataSourceProfile convert(
            DirectDataSourceProfile profile,
            Configuration conf); static final String KEY_PATH; static final String KEY_TEMP; static final String KEY_OUTPUT_STAGING; static final String KEY_OUTPUT_STREAMING; static final String KEY_MIN_FRAGMENT; static final String KEY_PREF_FRAGMENT; static final String KEY_SPLIT_BLOCKS; static final String KEY_COMBINE_BLOCKS; static final String KEY_KEEPALIVE_INTERVAL; static final String KEY_ROLLFORWARD_THREADS; static final String KEY_LEGACY_FRAGMENT_MIN; }