@Test(expected=IllegalArgumentException.class) public void testSetupNoParams() throws Exception { Configuration conf = new Configuration(); doReturn(conf).when(context).getConfiguration(); dispatcher.setup(context); }
@Override protected void setup(Context context) throws IOException, InterruptedException { String multipleOutputStr = context.getConfiguration().get(PROPERTY_MULTIPLEOUTPUTS); Preconditions.checkArgument(StringUtils.isNotBlank(multipleOutputStr), "required parameter '%s' is missing!", PROPERTY_MULTIPLEOUTPUTS); this.mos = instantiateMultipleOutputs(context); this.mimeTypeToPortNameMap = new HashMap<CharSequence, String>(); String[] portNames = StringUtils.split(context.getConfiguration().get(PROPERTY_MULTIPLEOUTPUTS)); for (String portName : portNames) { String currentMimeTypePropName = PROPERTY_PREFIX_MIMETYPES_CSV + portName; if (context.getConfiguration().get(currentMimeTypePropName) != null) { String[] currentPortMimeTypes = StringUtils.split( context.getConfiguration().get(currentMimeTypePropName), WorkflowRuntimeParameters.DEFAULT_CSV_DELIMITER); for (String currentPortMimeType : currentPortMimeTypes) { if (!currentPortMimeType.isEmpty() && !WorkflowRuntimeParameters.UNDEFINED_NONEMPTY_VALUE.equals(currentPortMimeType)) { this.mimeTypeToPortNameMap.put(currentPortMimeType.toLowerCase(), portName); } } } else { log.warn("undefined property '" + currentMimeTypePropName + "', no data will be dispatched to port '" + portName + "'"); } } }
DocumentContentUrlDispatcher extends Mapper<AvroKey<DocumentContentUrl>, NullWritable, NullWritable, NullWritable> { @Override protected void setup(Context context) throws IOException, InterruptedException { String multipleOutputStr = context.getConfiguration().get(PROPERTY_MULTIPLEOUTPUTS); Preconditions.checkArgument(StringUtils.isNotBlank(multipleOutputStr), "required parameter '%s' is missing!", PROPERTY_MULTIPLEOUTPUTS); this.mos = instantiateMultipleOutputs(context); this.mimeTypeToPortNameMap = new HashMap<CharSequence, String>(); String[] portNames = StringUtils.split(context.getConfiguration().get(PROPERTY_MULTIPLEOUTPUTS)); for (String portName : portNames) { String currentMimeTypePropName = PROPERTY_PREFIX_MIMETYPES_CSV + portName; if (context.getConfiguration().get(currentMimeTypePropName) != null) { String[] currentPortMimeTypes = StringUtils.split( context.getConfiguration().get(currentMimeTypePropName), WorkflowRuntimeParameters.DEFAULT_CSV_DELIMITER); for (String currentPortMimeType : currentPortMimeTypes) { if (!currentPortMimeType.isEmpty() && !WorkflowRuntimeParameters.UNDEFINED_NONEMPTY_VALUE.equals(currentPortMimeType)) { this.mimeTypeToPortNameMap.put(currentPortMimeType.toLowerCase(), portName); } } } else { log.warn("undefined property '" + currentMimeTypePropName + "', no data will be dispatched to port '" + portName + "'"); } } } }
DocumentContentUrlDispatcher extends Mapper<AvroKey<DocumentContentUrl>, NullWritable, NullWritable, NullWritable> { @Override protected void setup(Context context) throws IOException, InterruptedException { String multipleOutputStr = context.getConfiguration().get(PROPERTY_MULTIPLEOUTPUTS); Preconditions.checkArgument(StringUtils.isNotBlank(multipleOutputStr), "required parameter '%s' is missing!", PROPERTY_MULTIPLEOUTPUTS); this.mos = instantiateMultipleOutputs(context); this.mimeTypeToPortNameMap = new HashMap<CharSequence, String>(); String[] portNames = StringUtils.split(context.getConfiguration().get(PROPERTY_MULTIPLEOUTPUTS)); for (String portName : portNames) { String currentMimeTypePropName = PROPERTY_PREFIX_MIMETYPES_CSV + portName; if (context.getConfiguration().get(currentMimeTypePropName) != null) { String[] currentPortMimeTypes = StringUtils.split( context.getConfiguration().get(currentMimeTypePropName), WorkflowRuntimeParameters.DEFAULT_CSV_DELIMITER); for (String currentPortMimeType : currentPortMimeTypes) { if (!currentPortMimeType.isEmpty() && !WorkflowRuntimeParameters.UNDEFINED_NONEMPTY_VALUE.equals(currentPortMimeType)) { this.mimeTypeToPortNameMap.put(currentPortMimeType.toLowerCase(), portName); } } } else { log.warn("undefined property '" + currentMimeTypePropName + "', no data will be dispatched to port '" + portName + "'"); } } }  }
DocumentContentUrlDispatcher extends Mapper<AvroKey<DocumentContentUrl>, NullWritable, NullWritable, NullWritable> { @Override protected void setup(Context context) throws IOException, InterruptedException { String multipleOutputStr = context.getConfiguration().get(PROPERTY_MULTIPLEOUTPUTS); Preconditions.checkArgument(StringUtils.isNotBlank(multipleOutputStr), "required parameter '%s' is missing!", PROPERTY_MULTIPLEOUTPUTS); this.mos = instantiateMultipleOutputs(context); this.mimeTypeToPortNameMap = new HashMap<CharSequence, String>(); String[] portNames = StringUtils.split(context.getConfiguration().get(PROPERTY_MULTIPLEOUTPUTS)); for (String portName : portNames) { String currentMimeTypePropName = PROPERTY_PREFIX_MIMETYPES_CSV + portName; if (context.getConfiguration().get(currentMimeTypePropName) != null) { String[] currentPortMimeTypes = StringUtils.split( context.getConfiguration().get(currentMimeTypePropName), WorkflowRuntimeParameters.DEFAULT_CSV_DELIMITER); for (String currentPortMimeType : currentPortMimeTypes) { if (!currentPortMimeType.isEmpty() && !WorkflowRuntimeParameters.UNDEFINED_NONEMPTY_VALUE.equals(currentPortMimeType)) { this.mimeTypeToPortNameMap.put(currentPortMimeType.toLowerCase(), portName); } } } else { log.warn("undefined property '" + currentMimeTypePropName + "', no data will be dispatched to port '" + portName + "'"); } } }  @Override void cleanup(Context context); @Override void map(AvroKey<DocumentContentUrl> key, NullWritable ignore, Context context); }
DocumentContentUrlDispatcher extends Mapper<AvroKey<DocumentContentUrl>, NullWritable, NullWritable, NullWritable> { @Override protected void setup(Context context) throws IOException, InterruptedException { String multipleOutputStr = context.getConfiguration().get(PROPERTY_MULTIPLEOUTPUTS); Preconditions.checkArgument(StringUtils.isNotBlank(multipleOutputStr), "required parameter '%s' is missing!", PROPERTY_MULTIPLEOUTPUTS); this.mos = instantiateMultipleOutputs(context); this.mimeTypeToPortNameMap = new HashMap<CharSequence, String>(); String[] portNames = StringUtils.split(context.getConfiguration().get(PROPERTY_MULTIPLEOUTPUTS)); for (String portName : portNames) { String currentMimeTypePropName = PROPERTY_PREFIX_MIMETYPES_CSV + portName; if (context.getConfiguration().get(currentMimeTypePropName) != null) { String[] currentPortMimeTypes = StringUtils.split( context.getConfiguration().get(currentMimeTypePropName), WorkflowRuntimeParameters.DEFAULT_CSV_DELIMITER); for (String currentPortMimeType : currentPortMimeTypes) { if (!currentPortMimeType.isEmpty() && !WorkflowRuntimeParameters.UNDEFINED_NONEMPTY_VALUE.equals(currentPortMimeType)) { this.mimeTypeToPortNameMap.put(currentPortMimeType.toLowerCase(), portName); } } } else { log.warn("undefined property '" + currentMimeTypePropName + "', no data will be dispatched to port '" + portName + "'"); } } }  @Override void cleanup(Context context); @Override void map(AvroKey<DocumentContentUrl> key, NullWritable ignore, Context context);  }