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"); } for (int i = 0; i < numberOfAttributes + 1; ++i) { t.setField(new DefaultComparable(), i); } return TypeExtractor.getForObject(t); }  @Override void declare(final Fields fields); @Override void declare(final boolean direct, final Fields fields); @Override void declareStream(final String streamId, final Fields fields); @Override void declareStream(final String streamId, final boolean direct, final Fields fields);  }
@Test public void testNull() { Assert.assertNull(new FlinkOutputFieldsDeclarer().getOutputType(null)); }
@Test(expected = IllegalArgumentException.class) public void testUndeclared() { final FlinkOutputFieldsDeclarer declarer = new FlinkOutputFieldsDeclarer(); declarer.getOutputType("unknownStreamId"); }