class FlinkOutputFieldsDeclarer implements OutputFieldsDeclarer {

	TypeInformation<Tuple> getOutputType(final String streamId) throws IllegalArgumentException {
		if (streamId == null) {
			return null;
		}

		Fields outputSchema = this.outputStreams.get(streamId);
		if (outputSchema == null) {
			throw new IllegalArgumentException("Stream with ID '" + streamId
					+ "' was not declared.");
		}

		Tuple t;
		final int numberOfAttributes = outputSchema.size();

		if (numberOfAttributes <= 24) {
			try {
				t = Tuple.getTupleClass(numberOfAttributes + 1).newInstance();
			} catch (final InstantiationException e) {
				throw new RuntimeException(e);
			} catch (final IllegalAccessException e) {
				throw new RuntimeException(e);
			}
		} else {
			throw new IllegalArgumentException("Flink supports only a maximum number of 24 attributes");
		}

		// TODO: declare only key fields as DefaultComparable
		for (int i = 0; i < numberOfAttributes + 1; ++i) {
			t.setField(new DefaultComparable(), i);
		}

		return TypeExtractor.getForObject(t);
	}

}