@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 shouldBuildErrorWithKeyComponent() {
    // Given:
    final SerializationError<GenericRow> serError =
        new SerializationError<>(ERROR, Optional.of(RECORD), TOPIC, true);

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

    // Then:
    final Struct struct = (Struct) msg.value();
    final Struct serializationError = struct.getStruct(SERIALIZATION_ERROR);
    assertThat(
        serializationError.get(SERIALIZATION_ERROR_FIELD_TARGET),
        equalTo("key")
    );
  }