@Test(expected = NullPointerException.class) public void givenParametersWithoutInputClassPathResource_whenRunIsCalled_thenCheckFails() throws Exception { classPathResourceToHdfsCopier.run(mock(PortBindings.class), conf, new HashMap<>()); }
@Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkNotNull(parameters.get(PARAM_INPUT_CLASSPATH_RESOURCE), PARAM_INPUT_CLASSPATH_RESOURCE + " parameter was not specified!"); Preconditions.checkNotNull(parameters.get(PARAM_OUTPUT_HDFS_FILE_LOCATION), PARAM_OUTPUT_HDFS_FILE_LOCATION + " parameter was not specified!"); FileSystem fs = FileSystem.get(conf); try (InputStream in = classPathResourceProvider.apply(parameters.get(PARAM_INPUT_CLASSPATH_RESOURCE)); OutputStream os = fs.create(new Path(parameters.get(PARAM_OUTPUT_HDFS_FILE_LOCATION)))) { IOUtils.copyBytes(in, os, 4096, false); } }
ClassPathResourceToHdfsCopier implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkNotNull(parameters.get(PARAM_INPUT_CLASSPATH_RESOURCE), PARAM_INPUT_CLASSPATH_RESOURCE + " parameter was not specified!"); Preconditions.checkNotNull(parameters.get(PARAM_OUTPUT_HDFS_FILE_LOCATION), PARAM_OUTPUT_HDFS_FILE_LOCATION + " parameter was not specified!"); FileSystem fs = FileSystem.get(conf); try (InputStream in = classPathResourceProvider.apply(parameters.get(PARAM_INPUT_CLASSPATH_RESOURCE)); OutputStream os = fs.create(new Path(parameters.get(PARAM_OUTPUT_HDFS_FILE_LOCATION)))) { IOUtils.copyBytes(in, os, 4096, false); } } }
ClassPathResourceToHdfsCopier implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkNotNull(parameters.get(PARAM_INPUT_CLASSPATH_RESOURCE), PARAM_INPUT_CLASSPATH_RESOURCE + " parameter was not specified!"); Preconditions.checkNotNull(parameters.get(PARAM_OUTPUT_HDFS_FILE_LOCATION), PARAM_OUTPUT_HDFS_FILE_LOCATION + " parameter was not specified!"); FileSystem fs = FileSystem.get(conf); try (InputStream in = classPathResourceProvider.apply(parameters.get(PARAM_INPUT_CLASSPATH_RESOURCE)); OutputStream os = fs.create(new Path(parameters.get(PARAM_OUTPUT_HDFS_FILE_LOCATION)))) { IOUtils.copyBytes(in, os, 4096, false); } }  }
ClassPathResourceToHdfsCopier implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkNotNull(parameters.get(PARAM_INPUT_CLASSPATH_RESOURCE), PARAM_INPUT_CLASSPATH_RESOURCE + " parameter was not specified!"); Preconditions.checkNotNull(parameters.get(PARAM_OUTPUT_HDFS_FILE_LOCATION), PARAM_OUTPUT_HDFS_FILE_LOCATION + " parameter was not specified!"); FileSystem fs = FileSystem.get(conf); try (InputStream in = classPathResourceProvider.apply(parameters.get(PARAM_INPUT_CLASSPATH_RESOURCE)); OutputStream os = fs.create(new Path(parameters.get(PARAM_OUTPUT_HDFS_FILE_LOCATION)))) { IOUtils.copyBytes(in, os, 4096, false); } }  @Override void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters); @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts(); }
ClassPathResourceToHdfsCopier implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkNotNull(parameters.get(PARAM_INPUT_CLASSPATH_RESOURCE), PARAM_INPUT_CLASSPATH_RESOURCE + " parameter was not specified!"); Preconditions.checkNotNull(parameters.get(PARAM_OUTPUT_HDFS_FILE_LOCATION), PARAM_OUTPUT_HDFS_FILE_LOCATION + " parameter was not specified!"); FileSystem fs = FileSystem.get(conf); try (InputStream in = classPathResourceProvider.apply(parameters.get(PARAM_INPUT_CLASSPATH_RESOURCE)); OutputStream os = fs.create(new Path(parameters.get(PARAM_OUTPUT_HDFS_FILE_LOCATION)))) { IOUtils.copyBytes(in, os, 4096, false); } }  @Override void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters); @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts();  }