Propagating http headers in Kafka headers

[druminski]

Propagating HTTP headers as Kafka headers

Each allowed HTTP header is propagated as a separate Kafka header. Name of the Kafka header is prefix + HTTP header name. Prefix can be configured and its default value is h-. As an example, when Hermes is configured to propagate trace-id HTTP header, then it will be propagated via h-trace-id Kafka header. Hermes-consumers instead of reading HTTP headers from schema field __metadata, reads them from Kafka headers now.

Thanks to this change, we will be able to remove __metadata field from schema or make it non required as all necessary metadata are in Kafka headers now, read by hermes-consumers from them and propagated further as HTTP headers.

Additionally, this PR removes ts Kafka header from frontend module. Hermes in the past used this header to propagate timestamp of the message, but we don't need it anymore as Hermes uses dedicated and official timestamp field in the message record for it. It does that for a while, so we are save to stop propagating the ts header.

Configuration

New configuration

Frontend module

frontend.http.headers:
  propagation-as-kafka-headers:
    enabled: true/false #default is set to true
    prefix: 'h-' # 'h-' is a shortcut for header- in this context
  propagationEnabled: true/false # default is set to false
  allowedSet: '' # default is empty
  additionalAllowedSetToLog: '' #default is empty

Consumer module

consumer.http.headers.propagation-as-kafka-headers:
    enabled: true/false # default is set to true
    prefix: 'h-'

Removed configuration

Frontend module

frontend.header.propagation
  enabled: true/false # default was false
  allowFilter: '' # default was empty

Compatibility

When __metadata field in schema is present, then it is still filled with HTTP propagated headers by Hermes as it was previously and send to subscribers.

[moscicky]

Set#of creates an immutable collection so updating in updateInternalHeaders will throw an UnsupportedOperationException.

We should switch to mutable set here

or remove updateInternalHaders at all and implement isNotInternal as:

public boolean isNotInternal(String name) {
        return Set.of(this.messageId, this.timestamp, this.schemaId, this.schemaVersion).contains(name);
    }

or update the whole set on each field update

[moscicky]

ad1): the fact that it didn't blow up in integration tests means that tests for this part are missing

ad2): this would have been a compile time error in kotlin 😄

[druminski]

Nice catch. After your comment I made a combination of your ideas: when a field is updated then internalHeaders set is updated as well, but this time I initialize new immutable set. I decided to do it, because thanks to this when isNotInternal() is called (and it will be called for every message) then we don't have to initialize set and we still use immutable set.

[druminski]

As for the tests, it's a POJO configuration class for Kafka header names which we probably don't change, so I don't think that we have to tests fields modifications here. Important part is that it didn't break already existing tests.

[moscicky]

I agree that writing tests just for this class is not necessary. What I meant is that we have a piece of configuration that is never used is integration tests, so we never check if everything is wired correctly by spring and that properties are applied correctly. If we don't test all properties in e2e tests that we can only find out about bugs after deployment

[piotrrzysko]

Name of the Kafka header is the same as the HTTP header name.

I'm not sure if this is OK. If I'm not mistaken, it opens up the possibility of overwriting internal headers from outside of Hermes. I'd suggest at least giving higher priority to the internal headers or prefixing the external ones.

Also, I'm wondering about hiding this feature behind a flag. Currently, if someone doesn't want to migrate to this mechanism, they will still have to incur costs of storing additional data in Kafka.

[druminski]

Name of the Kafka header is the same as the HTTP header name.

I'm not sure if this is OK. If I'm not mistaken, it opens up the possibility of overwriting internal headers from outside of Hermes. I'd suggest at least giving higher priority to the internal headers or prefixing the external ones.

Also, I'm wondering about hiding this feature behind a flag. Currently, if someone doesn't want to migrate to this mechanism, they will still have to incur costs of storing additional data in Kafka.

@piotrrzysko I took into account your comment and:

1. introduced prefix for Kafka headers propagating HTTP headers
2. introduced flag which can enable/disable this propagation mechanism. True is its default value.
3. updated PR description
4. please take a look at the changes