UntypedValuesMergeFunction implements MergeFunction<MRecord, MRecord> { @Override public MRecord merge(MRecord currentValue) { if (!(currentValue instanceof UntypedValues)) { throw new InvalidParameterException(String.format( "ValuesMergeFunction currently only supports UntypedValues and not %s", currentValue.getClass().getName() )); } values = mapWithoutState(values, currentValue); return values; }  @Override MRecord merge(MRecord currentValue);  }
@Test public void merge() { mergeFunction.merge(record1); UntypedValues result = (UntypedValues) mergeFunction.merge(record2); assertEquals(123L, (long)result.getValue("test1").getAsLong()); assertEquals("string", result.getValue("test2").getAsString()); clone.merge(record1); result = (UntypedValues) clone.merge(record2); assertEquals(123L, (long)result.getValue("test1").getAsLong()); assertEquals("string", result.getValue("test2").getAsString()); }