FlinkOutputFieldsDeclarer implements OutputFieldsDeclarer { int[] getGroupingFieldIndexes(final String streamId, final List<String> groupingFields) { final int[] fieldIndexes = new int[groupingFields.size()]; for (int i = 0; i < fieldIndexes.length; ++i) { fieldIndexes[i] = this.outputStreams.get(streamId).fieldIndex(groupingFields.get(i)); } return fieldIndexes; }  @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 testGetGroupingFieldIndexes() { final int numberOfAttributes = 5 + this.r.nextInt(20); final String[] attributes = new String[numberOfAttributes]; for (int i = 0; i < numberOfAttributes; ++i) { attributes[i] = "a" + i; } final FlinkOutputFieldsDeclarer declarer = new FlinkOutputFieldsDeclarer(); declarer.declare(new Fields(attributes)); final int numberOfKeys = 1 + this.r.nextInt(24); final LinkedList<String> groupingFields = new LinkedList<String>(); final boolean[] indexes = new boolean[numberOfAttributes]; for (int i = 0; i < numberOfAttributes; ++i) { if (this.r.nextInt(25) < numberOfKeys) { groupingFields.add(attributes[i]); indexes[i] = true; } else { indexes[i] = false; } } final int[] expectedResult = new int[groupingFields.size()]; int j = 0; for (int i = 0; i < numberOfAttributes; ++i) { if (indexes[i]) { expectedResult[j++] = i; } } final int[] result = declarer.getGroupingFieldIndexes(Utils.DEFAULT_STREAM_ID, groupingFields); Assert.assertEquals(expectedResult.length, result.length); for (int i = 0; i < expectedResult.length; ++i) { Assert.assertEquals(expectedResult[i], result[i]); } }