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

    return new SchemaAndValue(ProcessingLogMessageSchema.PROCESSING_LOG_SCHEMA, struct);
  }
@Test
  public void shouldBuildDeserializationError() {
    // Given:
    final DeserializationError deserError = new DeserializationError(
        error,
        Optional.of(record),
        "topic",
        false
    );

    // When:
    final SchemaAndValue msg = deserError.get(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(MessageType.DESERIALIZATION_ERROR.getTypeId()));
    final Struct deserializationError = struct.getStruct(DESERIALIZATION_ERROR);
    assertThat(
        deserializationError.get(DESERIALIZATION_ERROR_FIELD_TARGET),
        equalTo("value")
    );
    assertThat(
        deserializationError.get(DESERIALIZATION_ERROR_FIELD_MESSAGE),
        equalTo(error.getMessage())
    );
    assertThat(
        deserializationError.get(DESERIALIZATION_ERROR_FIELD_CAUSE),
        equalTo(causeList)
    );
    assertThat(
        deserializationError.get(DESERIALIZATION_ERROR_FIELD_RECORD_B64),
        equalTo(Base64.getEncoder().encodeToString(record))
    );
    assertThat(
        deserializationError.get(DESERIALIZATION_ERROR_FIELD_TOPIC),
        equalTo("topic")
    );
    schema.fields().forEach(
        f -> {
          if (!ImmutableList.of(TYPE, DESERIALIZATION_ERROR).contains(f.name())) {
            assertThat(struct.get(f), is(nullValue()));
          }
        }
    );
  }