### Question:
ExplodeFilter extends AbstractMergeRecordFilter<ExplodeFilter> { @Override protected RecordsIterable<TypedStruct> apply(final FilterContext context, final TypedStruct record) throws FilterException { final TypedValue value = checkIsNotNull(record.find(config.source())); if (value.type() != Type.ARRAY) { throw new FilterException( "Invalid type for field '" + config.source() + "', expected ARRAY, was " + value.type()); } final List<TypedStruct> explode = value.getArray() .stream() .map(it -> TypedStruct.create().insert(config.source(), TypedValue.any(it))) .collect(Collectors.toList()); return new RecordsIterable<>(explode); }  @Override void configure(final Map<String, ?> props); @Override ConfigDef configDef();  }### Answer:
@Test public void should_explode_input_given_array_of_primitive() { RecordsIterable<TypedStruct> result = filter.apply(null, create().put("message", PRIMITIVE_VALUES), false); Assert.assertEquals(PRIMITIVE_VALUES.size(), result.size()); Assert.assertEquals(PRIMITIVE_VALUES, extractStringValues(result, it -> it.getString("message"))); }
@Test public void should_explode_input_given_array_of_complex() { RecordsIterable<TypedStruct> result = filter.apply(null, create().put("message", COMPLEXES_VALUES), false); Assert.assertEquals(COMPLEXES_VALUES.size(), result.size()); Assert.assertEquals(COMPLEXES_VALUES, extractStringValues(result, it -> it.getStruct("message"))); }