@Test( expected = IllegalArgumentException.class ) public void shouldNotBuildBuilderForOperatorWithExceedingOutputPortSchema () { OperatorDefBuilder.newInstance( "op1", OperatorWithExceedingOutputPortSchema.class ); }
public static OperatorDefBuilder newInstance ( final String id, final Class<? extends Operator> clazz ) { failIfEmptyOperatorId( id ); checkArgument( clazz != null, "clazz must be provided" ); final OperatorSpec spec = getOperatorSpecOrFail( clazz ); final OperatorSchema schema = getOperatorSchema( clazz ); if ( schema != null ) { if ( schema.inputs().length > 0 ) { checkArgument( spec.inputPortCount() != DYNAMIC_PORT_COUNT, "input port count must be defined in OperatorSpec if operator is annotated with OperatorSchema" ); checkArgument( spec.inputPortCount() >= schema.inputs().length, "Number of input port schemas in OperatorSchema exceeds input port count in OperatorSpec" ); } if ( schema.outputs().length > 0 ) { checkArgument( spec.outputPortCount() != DYNAMIC_PORT_COUNT, "output port count must be defined in OperatorSpec if operator is annotated with OperatorSchema" ); checkArgument( spec.outputPortCount() >= schema.outputs().length, "Number of output port schemas in OperatorSchema exceeds output port count in OperatorSpec" ); } failIfOperatorSchemaHasDuplicatePortIndices( schema ); failIfInvalidPortIndexOnPortSchemas( spec.inputPortCount(), schema.inputs() ); failIfInvalidPortIndexOnPortSchemas( spec.outputPortCount(), schema.outputs() ); } if ( spec.inputPortCount() != DYNAMIC_PORT_COUNT ) { failIfInvalidPortCount( spec.type(), spec.inputPortCount(), "input" ); failIfStatelessOperatorWithMultipleInputPorts( spec.type(), spec.inputPortCount(), "input" ); } if ( spec.outputPortCount() != DYNAMIC_PORT_COUNT ) { failIfInvalidPortCount( spec.type(), spec.outputPortCount(), "output" ); failIfStatelessOperatorWithNoOutputPorts( spec.type(), spec.outputPortCount(), "output" ); } return new OperatorDefBuilder( id, clazz, spec, schema ); }
OperatorDefBuilder { public static OperatorDefBuilder newInstance ( final String id, final Class<? extends Operator> clazz ) { failIfEmptyOperatorId( id ); checkArgument( clazz != null, "clazz must be provided" ); final OperatorSpec spec = getOperatorSpecOrFail( clazz ); final OperatorSchema schema = getOperatorSchema( clazz ); if ( schema != null ) { if ( schema.inputs().length > 0 ) { checkArgument( spec.inputPortCount() != DYNAMIC_PORT_COUNT, "input port count must be defined in OperatorSpec if operator is annotated with OperatorSchema" ); checkArgument( spec.inputPortCount() >= schema.inputs().length, "Number of input port schemas in OperatorSchema exceeds input port count in OperatorSpec" ); } if ( schema.outputs().length > 0 ) { checkArgument( spec.outputPortCount() != DYNAMIC_PORT_COUNT, "output port count must be defined in OperatorSpec if operator is annotated with OperatorSchema" ); checkArgument( spec.outputPortCount() >= schema.outputs().length, "Number of output port schemas in OperatorSchema exceeds output port count in OperatorSpec" ); } failIfOperatorSchemaHasDuplicatePortIndices( schema ); failIfInvalidPortIndexOnPortSchemas( spec.inputPortCount(), schema.inputs() ); failIfInvalidPortIndexOnPortSchemas( spec.outputPortCount(), schema.outputs() ); } if ( spec.inputPortCount() != DYNAMIC_PORT_COUNT ) { failIfInvalidPortCount( spec.type(), spec.inputPortCount(), "input" ); failIfStatelessOperatorWithMultipleInputPorts( spec.type(), spec.inputPortCount(), "input" ); } if ( spec.outputPortCount() != DYNAMIC_PORT_COUNT ) { failIfInvalidPortCount( spec.type(), spec.outputPortCount(), "output" ); failIfStatelessOperatorWithNoOutputPorts( spec.type(), spec.outputPortCount(), "output" ); } return new OperatorDefBuilder( id, clazz, spec, schema ); } }
OperatorDefBuilder { public static OperatorDefBuilder newInstance ( final String id, final Class<? extends Operator> clazz ) { failIfEmptyOperatorId( id ); checkArgument( clazz != null, "clazz must be provided" ); final OperatorSpec spec = getOperatorSpecOrFail( clazz ); final OperatorSchema schema = getOperatorSchema( clazz ); if ( schema != null ) { if ( schema.inputs().length > 0 ) { checkArgument( spec.inputPortCount() != DYNAMIC_PORT_COUNT, "input port count must be defined in OperatorSpec if operator is annotated with OperatorSchema" ); checkArgument( spec.inputPortCount() >= schema.inputs().length, "Number of input port schemas in OperatorSchema exceeds input port count in OperatorSpec" ); } if ( schema.outputs().length > 0 ) { checkArgument( spec.outputPortCount() != DYNAMIC_PORT_COUNT, "output port count must be defined in OperatorSpec if operator is annotated with OperatorSchema" ); checkArgument( spec.outputPortCount() >= schema.outputs().length, "Number of output port schemas in OperatorSchema exceeds output port count in OperatorSpec" ); } failIfOperatorSchemaHasDuplicatePortIndices( schema ); failIfInvalidPortIndexOnPortSchemas( spec.inputPortCount(), schema.inputs() ); failIfInvalidPortIndexOnPortSchemas( spec.outputPortCount(), schema.outputs() ); } if ( spec.inputPortCount() != DYNAMIC_PORT_COUNT ) { failIfInvalidPortCount( spec.type(), spec.inputPortCount(), "input" ); failIfStatelessOperatorWithMultipleInputPorts( spec.type(), spec.inputPortCount(), "input" ); } if ( spec.outputPortCount() != DYNAMIC_PORT_COUNT ) { failIfInvalidPortCount( spec.type(), spec.outputPortCount(), "output" ); failIfStatelessOperatorWithNoOutputPorts( spec.type(), spec.outputPortCount(), "output" ); } return new OperatorDefBuilder( id, clazz, spec, schema ); } private  OperatorDefBuilder( final String id,
                                 final Class<? extends Operator> clazz,
                                 final OperatorSpec spec,
                                 final OperatorSchema schema ); }
OperatorDefBuilder { public static OperatorDefBuilder newInstance ( final String id, final Class<? extends Operator> clazz ) { failIfEmptyOperatorId( id ); checkArgument( clazz != null, "clazz must be provided" ); final OperatorSpec spec = getOperatorSpecOrFail( clazz ); final OperatorSchema schema = getOperatorSchema( clazz ); if ( schema != null ) { if ( schema.inputs().length > 0 ) { checkArgument( spec.inputPortCount() != DYNAMIC_PORT_COUNT, "input port count must be defined in OperatorSpec if operator is annotated with OperatorSchema" ); checkArgument( spec.inputPortCount() >= schema.inputs().length, "Number of input port schemas in OperatorSchema exceeds input port count in OperatorSpec" ); } if ( schema.outputs().length > 0 ) { checkArgument( spec.outputPortCount() != DYNAMIC_PORT_COUNT, "output port count must be defined in OperatorSpec if operator is annotated with OperatorSchema" ); checkArgument( spec.outputPortCount() >= schema.outputs().length, "Number of output port schemas in OperatorSchema exceeds output port count in OperatorSpec" ); } failIfOperatorSchemaHasDuplicatePortIndices( schema ); failIfInvalidPortIndexOnPortSchemas( spec.inputPortCount(), schema.inputs() ); failIfInvalidPortIndexOnPortSchemas( spec.outputPortCount(), schema.outputs() ); } if ( spec.inputPortCount() != DYNAMIC_PORT_COUNT ) { failIfInvalidPortCount( spec.type(), spec.inputPortCount(), "input" ); failIfStatelessOperatorWithMultipleInputPorts( spec.type(), spec.inputPortCount(), "input" ); } if ( spec.outputPortCount() != DYNAMIC_PORT_COUNT ) { failIfInvalidPortCount( spec.type(), spec.outputPortCount(), "output" ); failIfStatelessOperatorWithNoOutputPorts( spec.type(), spec.outputPortCount(), "output" ); } return new OperatorDefBuilder( id, clazz, spec, schema ); } private  OperatorDefBuilder( final String id,
                                 final Class<? extends Operator> clazz,
                                 final OperatorSpec spec,
                                 final OperatorSchema schema ); static OperatorDefBuilder newInstance( final String id, final Class<? extends Operator> clazz ); OperatorDefBuilder setInputPortCount( final int inputPortCount ); OperatorDefBuilder setOutputPortCount( final int outputPortCount ); OperatorDefBuilder setConfig( final OperatorConfig config ); OperatorDefBuilder setExtendingSchema( final OperatorRuntimeSchemaBuilder extendingSchemaBuilder ); OperatorDefBuilder setExtendingSchema( final OperatorRuntimeSchema extendingSchema ); OperatorDefBuilder setPartitionFieldNames( final List<String> partitionFieldNames ); OperatorDef build(); }
OperatorDefBuilder { public static OperatorDefBuilder newInstance ( final String id, final Class<? extends Operator> clazz ) { failIfEmptyOperatorId( id ); checkArgument( clazz != null, "clazz must be provided" ); final OperatorSpec spec = getOperatorSpecOrFail( clazz ); final OperatorSchema schema = getOperatorSchema( clazz ); if ( schema != null ) { if ( schema.inputs().length > 0 ) { checkArgument( spec.inputPortCount() != DYNAMIC_PORT_COUNT, "input port count must be defined in OperatorSpec if operator is annotated with OperatorSchema" ); checkArgument( spec.inputPortCount() >= schema.inputs().length, "Number of input port schemas in OperatorSchema exceeds input port count in OperatorSpec" ); } if ( schema.outputs().length > 0 ) { checkArgument( spec.outputPortCount() != DYNAMIC_PORT_COUNT, "output port count must be defined in OperatorSpec if operator is annotated with OperatorSchema" ); checkArgument( spec.outputPortCount() >= schema.outputs().length, "Number of output port schemas in OperatorSchema exceeds output port count in OperatorSpec" ); } failIfOperatorSchemaHasDuplicatePortIndices( schema ); failIfInvalidPortIndexOnPortSchemas( spec.inputPortCount(), schema.inputs() ); failIfInvalidPortIndexOnPortSchemas( spec.outputPortCount(), schema.outputs() ); } if ( spec.inputPortCount() != DYNAMIC_PORT_COUNT ) { failIfInvalidPortCount( spec.type(), spec.inputPortCount(), "input" ); failIfStatelessOperatorWithMultipleInputPorts( spec.type(), spec.inputPortCount(), "input" ); } if ( spec.outputPortCount() != DYNAMIC_PORT_COUNT ) { failIfInvalidPortCount( spec.type(), spec.outputPortCount(), "output" ); failIfStatelessOperatorWithNoOutputPorts( spec.type(), spec.outputPortCount(), "output" ); } return new OperatorDefBuilder( id, clazz, spec, schema ); } private  OperatorDefBuilder( final String id,
                                 final Class<? extends Operator> clazz,
                                 final OperatorSpec spec,
                                 final OperatorSchema schema ); static OperatorDefBuilder newInstance( final String id, final Class<? extends Operator> clazz ); OperatorDefBuilder setInputPortCount( final int inputPortCount ); OperatorDefBuilder setOutputPortCount( final int outputPortCount ); OperatorDefBuilder setConfig( final OperatorConfig config ); OperatorDefBuilder setExtendingSchema( final OperatorRuntimeSchemaBuilder extendingSchemaBuilder ); OperatorDefBuilder setExtendingSchema( final OperatorRuntimeSchema extendingSchema ); OperatorDefBuilder setPartitionFieldNames( final List<String> partitionFieldNames ); OperatorDef build();  }