@Override
  public SchemaAndValue get(final ProcessingLogConfig config) {
    final Struct struct = new Struct(ProcessingLogMessageSchema.PROCESSING_LOG_SCHEMA)
        .put(ProcessingLogMessageSchema.TYPE, MessageType.SERIALIZATION_ERROR.getTypeId())
        .put(ProcessingLogMessageSchema.SERIALIZATION_ERROR, serializationError(config));

    return new SchemaAndValue(ProcessingLogMessageSchema.PROCESSING_LOG_SCHEMA, struct);
  }
@Test
  public void shouldBuildSerializationError() {
    // Given:
    final SerializationError<GenericRow> serError = new SerializationError<>(
        ERROR,
        Optional.of(RECORD),
        TOPIC,
        false
    );

    // When:
    final SchemaAndValue msg = serError.get(LOGGING_CONFIG);

    // Then:
    final Schema schema = msg.schema();
    assertThat(schema, equalTo(PROCESSING_LOG_SCHEMA));
    final Struct struct = (Struct) msg.value();
    assertThat(
        struct.get(ProcessingLogMessageSchema.TYPE),
        equalTo(ProcessingLogMessageSchema.MessageType.SERIALIZATION_ERROR.getTypeId()));
    final Struct serializationError = struct.getStruct(SERIALIZATION_ERROR);
    assertThat(
        serializationError.get(SERIALIZATION_ERROR_FIELD_TARGET),
        equalTo("value")
    );
    assertThat(
        serializationError.get(SERIALIZATION_ERROR_FIELD_MESSAGE),
        equalTo(ERROR.getMessage())
    );
    assertThat(
        serializationError.get(SERIALIZATION_ERROR_FIELD_CAUSE),
        equalTo(CAUSE_LIST)
    );
    assertThat(
        serializationError.get(SERIALIZATION_ERROR_FIELD_RECORD),
        equalTo(RECORD.toString())
    );
    assertThat(
        serializationError.get(SERIALIZATION_ERROR_FIELD_TOPIC),
        equalTo(TOPIC)
    );
    schema.fields().forEach(
        f -> {
          if (!ImmutableList.of(TYPE, SERIALIZATION_ERROR).contains(f.name())) {
            assertThat(struct.get(f), is(nullValue()));
          }
        }
    );
  }