### Question:
JSONFilter extends AbstractMergeRecordFilter<JSONFilter> { @Override protected RecordsIterable<TypedStruct> apply(final FilterContext context, final TypedStruct record) { final String value = extractJsonField(checkIsNotNull(record.get(configs.source()))); final TypedValue typedValue; try { typedValue = converter.readJson(value); } catch (Exception e) { throw new FilterException(e.getLocalizedMessage(), e.getCause()); } final Type type = typedValue.type(); if (type != Type.ARRAY && type != Type.STRUCT) { throw new FilterException( "Cannot process JSON value with unsupported type. Expected Array or Object, was " + type); } if (type == Type.STRUCT && configs.merge()) { return RecordsIterable.of(typedValue.getStruct()); } if (type == Type.ARRAY) { if (configs.explode()) { Collection<?> items = typedValue.getArray(); ArraySchema arraySchema = (ArraySchema)typedValue.schema(); Type arrayValueType = arraySchema.valueSchema().type(); if (configs.merge()) { if (arrayValueType == Type.STRUCT) { final List<TypedStruct> records = items .stream() .map(it -> TypedValue.any(it).getStruct()) .collect(Collectors.toList()); return new RecordsIterable<>(records); } throw new FilterException( "Unsupported operation. Cannot merge array value of type '" + arrayValueType + "' into the top level of the input record"); } final List<TypedStruct> records = items .stream() .map(it -> TypedStruct.create().put(targetField(), TypedValue.of(it, arrayValueType))) .collect(Collectors.toList()); return new RecordsIterable<>(records); } if (configs.merge()) { throw new FilterException( "Unsupported operation. Cannot merge JSON Array into the top level of the input record"); } } return RecordsIterable.of(TypedStruct.create().put(targetField(), typedValue)); }  @Override void configure(final Map<String, ?> props); @Override ConfigDef configDef();  }### Answer:
@Test public void should_add_parsed_json_into_source_given_no_target_field() { List<TypedStruct> expected = Collections.singletonList(create() .put("message", create().put("firstName", "foo").put("lastName", "bar")) ); assertOutput(filter.apply(null, STRING_RECORD, false).collect(), expected); }
@Test public void should_parse_input_given_value_of_type_bytes() { List<TypedStruct> expected = Collections.singletonList(create() .put("message", create().put("firstName", "foo").put("lastName", "bar")) ); assertOutput(filter.apply(null, BYTES_RECORD, false).collect(), expected); }