Address content-type compatibility between messages generated by different versions of SCSt apps

[olegz]

Due to the recent content-type negotiation improvements we unintentionally introduced compatibility issue where pre-2.0 consumers may not be able to properly negotiate the content type of the message generated by 2.0+ producers.
This is due to the fact that pre-2.0 producers and consumers were relying on the original-content-type header which was removed in 2.0.
Another change that was introduced as part of this effort deals with registering additional flag (spring.cloud.stream.bindings.input.legacyContentTypeHeaderEnabled) and interceptor (LegacyContentTypeHeaderInterceptor) to allow 2.0+ consumers to successfully process pre-2.0 messages. This issue is that it makes consumer statically aware of the producer and its version and could result in more issues if such consumer receives messages from the producers of different versions.

Proposal:

1. 2.0 producer should re-introduce the original-content-type header. While this header will be ignored by 2.0 consumers it will be used by pre-2.0 consumers as before. It should also generate one additional header such as version (e.g., version=2.x). This additional header will be used by 2.0+ consumers to ignore the re-introduced original-content-type header (more details below).
2. 2.0 consumer should have a mechanism (may be through the already existing LegacyContentTypeHeaderInterceptor) to check for the version header before checking for the existence of the original-content-type header. If 'version' header is pre-2.0 or is not present (which is what's going to be in the case of pre-2.0 producers) then we know it's a legacy message and content-type negotiation should be done using the original-content-type header, otherwise such header will be ignored. Basically the version header will play the same role as newly introduced spring.cloud.stream.bindings.input.legacyContentTypeHeaderEnabled flag, but it will come as part of the message instead of static configuration of the consumer. Further more, making it a version instead of a boolean will give us more flexibility in the future when it comes to similar issues especially if we decide to drop support for certain version of messages.

[garyrussell]

What about Kafka? Does this imply that a 2.0 Kafka consumer has to interoperate with a 1.x producer (and vice-versa)?

We made the decision that 2.0 would only support >= Kafka 0.11; it sounds like this won't be possible now and we'll have to maintain support for the 0.10.x binder(s).

Of course, transactions and native headers won't be available if you use such a binder.

It will also be difficult to to eventually migrate to kafka 0.11 (or 1.0 when available) because it will need a big bang. To support a progressive migration, we'll need some work in the binder to not add native headers (so the 0.11 binder can talk to an 0.10 broker). When all services have migrated to 0.11 binder; they could then change out the broker; they would still then need a "big bang" to enable native headers across all the services.

Documenting all this is going to be difficult.

Sigh.

[sobychacko]

Since we only support Kafka 0.11 and up in 2.0, how about we say that in the case of Kafka based apps, 1.3 is backward/forward compatible with 2.0, as long as it uses the latest Kafka broker (since 1.3 can work with 0.11 by ignoring native headers and transactions). That way, we don't have to bring support for all the 0.10.x versions of Kafka in 2.0 which is going to be very challenging.

[garyrussell]

(since 1.3 can work with 0.11 by ignoring native headers and transactions)

But that's not the whole picture; when using 1.3 with the 0.11 binder it does add headers. If you use 1.3 with a 0.10 binder, it doesn't use headers.

2.0 will not be able to produce messages for a 1.3/0.10 consumer, unless we add code to skip headers and reinstate support for embedded headers.

[markpollack]

Was a design where a 2.0 producer didn't add this legacy field considered? The downside of this approach is dragging the legacy fields along which keeps potential confusion. While somewhat less comprehensive, an approach in which the latest 1.3.x release is made to be forward compatible, by being aware of where the new 2.0 fields/headers are located. I know it is complicated, but just wanted to present another approach in case it wasn't considered.

[olegz]

@markpollack that can work, just need to wrap my head around it. And it will require a new release of 1.3 pipeline (i.e., 1.3.1).
That said, Gary's Kafka points are something we need to take great caution, since it brings another variable into the picture (ability to support message headers natively). Here are few scenarios we should consider:

1. Kafka 0.11 producer -> (0.11 broker) -> Kafka 0.10 consumer

2. Kafka 0.10 producer -> (0.11 broker) -> Kafka 0.11 consumer
   . . . however in the 1 scenario the consumer won't be able to understand the message due native headers. In the 2 scenario, we can make consumer understand that headers are embedded in payload. However, there may/will be an issue with non-SCSt consumers for this scenarios as they may expect headers to be available natively and I am not sure we can do it with Kafka 0.10 client.

3. Kafka 0.11 producer -> (0.10 broker) -> Kafka 0.10 consumer

4. Kafka 0.10 producer -> (0.10 broker) -> Kafka 0.11 consumer
   The 3 may work if there is a way for the producer to know which broker version it's bound to, so it can determine how to deal with headers (embed or native). The 4 will work the same way as 2, however it will not have the non-SCSt client problem.

We obviously should not have problems when both producers, consumers and brokers are all of a same version and since 1.3 dropped support for pre 0.10 Kafka we don't have to worry about older brokers/clients.

[garyrussell]

I haven't (yet) found a way for the client to figure out the broker version (only the kafka-clients version).

In any case, even if we can figure out the broker version (say it's 0.11) the producer doesn't know whether the consumer is using the 0.11 client, so we can't make any assumptions based on the broker version anyway.

I have issued a PR adding a new HeaderMode (headers). This is a switch which will allow 2.0 producers/consumers to coexist with 1.x apps (albeit via configuration). The kafka binder will continue to default to embedded headers, thus easing migration. When all (associated) apps are 2.x, the user can reconfigure them to use the native headers; the 2.0 consumer should check for native headers and, if not present, see if we have embedded headers and, if so, extract them. We can potentially remove this fallback logic in 2.1 or 3.0.

I think this covers all the bases.

[olegz]

The only bases it is not covering is non-SCSt clients (producers/consumers), but as we discussed that level of the awareness may need to be delegated to such clients. Correct?

[garyrussell]

Most non-scst peers will be ok.

headerMode=none (P/C) when communicating with 0.10 peers (or 0.11 peers that don't populate or expect headers); no problems.

headerMode=headers (C) when communicating with 0.11 producers that populate headers - if the default header mapper detects the headers were'n't created by S-I-Kafka, it will return the raw headers; no problems.

headerMode=headers (P) when communicating with 0.11 consumers - a consumer expecting headers will either need to be made aware of the S-I-K header encoding, or the producer will need to be configured with a different header mapper implementation - we don't currently expose a mechanism for that, so we will need to.

[artembilan]

Reopen.
Closed automatically by merge relevant PR