class GafferMapFunction implements FlatMapFunction<T, Element> {

    @Override
    public void flatMap(final T item, final Collector<Element> out) throws Exception {
        if (null == out) {
            throw new IllegalArgumentException("Element collector is required");
        }

        if (null == elementGenerator) {
            elementGenerator = generatorClassName.newInstance();
        }

        if (elementGenerator instanceof OneToOneElementGenerator) {
            out.collect(((OneToOneElementGenerator<T>) elementGenerator)._apply(item));
        } else if (elementGenerator instanceof OneToManyElementGenerator) {
            final Iterable<Element> elements = ((OneToManyElementGenerator<T>) elementGenerator)._apply(item);
            if (null != elements) {
                elements.forEach(out::collect);
            }
        } else {
            final Iterable<? extends Element> elements = (elementGenerator).apply(Collections.singleton(item));
            if (null != elements) {
                elements.forEach(out::collect);
            }
        }
    }

    public  GafferMapFunction();
    public  GafferMapFunction(final Class<T> consumeAs, final Class<? extends Function<Iterable<? extends T>, Iterable<? extends Element>>> generatorClassName);

}