@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 shouldSetNullRecordToNull() {
    // Given:
    final DeserializationError deserError = new DeserializationError(
        error,
        Optional.empty(),
        "topic",
        false
    );

    // When:
    final SchemaAndValue msg = deserError.get(config);

    // Then:
    final Struct struct = (Struct) msg.value();
    final Struct deserializationError = struct.getStruct(DESERIALIZATION_ERROR);
    assertThat(deserializationError.get(DESERIALIZATION_ERROR_FIELD_RECORD_B64),
        is(nullValue()));
  }