class StaticTopicSerde implements Serde<T> {

  @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;
      }
    };
  }

  private  StaticTopicSerde(
      final String topic,
      final Serde<T> delegate,
      final Callback onFailure
  );

}