@Override
  public Deserializer<T> deserializer() {
    final Deserializer<T> deserializer = delegate.deserializer();

    if (deserializer instanceof LoggingDeserializer<?>) {
      final LoggingDeserializer<T> loggingDeserializer = (LoggingDeserializer<T>) deserializer;

      return (topic, data) -> {
        final DelayedResult<T> staticResult = loggingDeserializer.tryDeserialize(this.topic, data);
        if (!staticResult.isError()) {
          return staticResult.get();
        }

        // if both attempts error, then staticResult.get() will log the error to
        // the processing log and throw - do not call the callback in this case
        final DelayedResult<T> sourceResult = loggingDeserializer.tryDeserialize(topic, data);
        if (sourceResult.isError()) {
          return staticResult.get();
        }

        onFailure.onDeserializationFailure(topic, this.topic, data);
        return sourceResult.get();
      };
    }

    return (topic, data) -> {
      try {
        return deserializer.deserialize(this.topic, data);
      } catch (final Exception e) {
        final T object = deserializer.deserialize(topic, data);
        onFailure.onDeserializationFailure(topic, this.topic, data);
        return object;
      }
    };
  }
@Test
  public void shouldUseDelegateDeserializerWithStaticTopic() {
    // When:
    final Object deserialized = staticSerde.deserializer().deserialize(SOURCE_TOPIC, SOME_BYTES);

    // Then:
    verify(delegateD).deserialize(STATIC_TOPIC, SOME_BYTES);
    assertThat(deserialized, is(SOME_OBJECT));
    verifyNoMoreInteractions(callback);
  }