KAFKA-13769: Explicitly route FK join results to correct partitions

[Gerrrr]

Prior to this commit FK response sink routed FK results to
SubscriptionResolverJoinProcessorSupplier using the primary key.

There are cases, where this behaviour is incorrect. For example,
if KTable key serde differs from the data source serde which might
happen without a key changing operation.

Instead of determining the resolver partition by serializing the PK
this patch includes target partition in SubscriptionWrapper and
SubscriptionResponseWrapper payloads. Default FK response-sink
partitioner extracts the correct partition from the value and
routes the message accordingly.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[mjsax]

Can we unify the dataLength computation into a single place? It's a little bit scattered at hard to follow. Also with different versions, I like a more explicit pattern using switch:

switch (version) {
case 0: encodeV1(); break;
case 1: encodeV2(); break;
default: throw...
}

[Gerrrr]

Done in a59a4f6

[mjsax]

I think we cannot just write the new format blindly. The receiving task may be executed on a different instance that might not have been ungraded yet during a rolling upgrade. -- We can only start to use the new format after all instances are upgraded, otherwise older instances would crash (or the upgrade can not be done in rolling bounce fashion what seen not acceptable).

Not sure right now how we can handle this case though: The coarse grained solution would be to use a config and let users do two rolling bounces (requires a KIP). Or we try to piggy-back on the rebalance metadata version, but I am not sure how easy it would be to do, given that we setup the serdes at a totally different layer and get access to the rebalance metadata version only during runtime.

[vvcephei]

Yeah, this is indeed a problem. It's the same issue I documented here: https://issues.apache.org/jira/browse/KAFKA-10336

I was also thinking that the rebalance metadata version could give us a mechanism to coordinate this kind of thing across a rolling bounce. I documented the approach in that ticket, but never got a chance to work on it.

If we don't want to bite off a general solution to it, two rolling bounces might be the best compromise, as annoying as they are.

Alternatively, since we actually do check the version and throw an exception if we encounter it, it is "safe" (in that there's no data corruption) to just go ahead and let the exceptions happen during a rolling bounce. Not the nicest thing to do, though.

[Gerrrr]

What will happen when an old processor receives a v1 record and throws an exception? Will it throw that record on the floor or just blocks processing until the upgrade?

As John suggested in KAFKA-10336, we can pass UPGRADE_FROM into the serde so that it produces v0 records if the option is set. I will also add 3.3 as a valid choice for that option. WDYT?

[mjsax]

What will happen when an old processor receives a v1 record and throws an exception? Will it throw that record on the floor or just blocks processing until the upgrade?

The thread with crash, and we will eventually rebalance. So there is no data-loss, but for at-least-once duplicate processing. It's very bad user experience if they upgrade one instance and suddenly all other might crash...

Re-using UPGRAD_FROM should work. The config is already there, so maybe it's ok if we just add a never "accepted value" without doing a KIP? We need to cover this case in the upgrade docs though!

[Gerrrr]

Added UPGRADE_FROM in d9f6c52. There, I added all versions from 2.7 to 3.2 as UPGRADE_FROM options. If serde detects the upgrade, it uses v0 format to serialize the data. Deserializer does not need any changes as there is no difference between v0 and upgrade-from-v0 formats.

[mjsax]

Can we move this to the end of the byte array and thus re-use serializeV0 to write the prefix?

[Gerrrr]

I am not sure if this is a good idea. As an advantage, we will remove duplicated parts in serialize. However, there are 2 downsides:

• We use v0 path in 2 situations - when it is real v0 and when we upgrade from v0 (i.e. a data record can actually have a primary partition and v1 as part of its state). This means that we can't just put data.getVersion() into the buffer, but have to hardcode it to 0. Making this logic generic enough to handle v0, upgrade-from, and v1 will be cumbersome.
• Moving a fixed-size field to the end of the byte array will complicate deserialization. Consider

kafka/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionWrapperSerde.java

Lines 234 to 242 in d9f6c52

	final byte[] primaryKeyRaw = new byte[data.length - lengthSum]; //The remaining data is the serialized pk
	buf.get(primaryKeyRaw, 0, primaryKeyRaw.length);
	if (primaryKeySerializationPseudoTopic == null) {
	primaryKeySerializationPseudoTopic = primaryKeySerializationPseudoTopicSupplier.get();
	}
	final K primaryKey = primaryKeyDeserializer.deserialize(primaryKeySerializationPseudoTopic,
	primaryKeyRaw);

Right now, the logic for finding PK size is straightforward. It can be summarized as "take it from here and until the end of the buffer". If we move primary partition to the end, we'll have to remove another 4 bytes iff the version is greater than 0.

I'd say that code duplication is lesser evil in this case. WDYT? I am happy to move primary partition to the end if you don't agree.

[mjsax]

This means that we can't just put data.getVersion() into the buffer, but have to hardcode it to 0

Yes, but this seems ok? Ie, we would have a serializeV0Internal(final short version) that is called as (something like):

public byte[] serializeV0() {
  return serializeV0Internal(0).getBytes();
}

public byte[] serializeV1() {
  final ByteStream out = serializeV0Internal(0);
  
  // add v1 stuff to `out`
  
  return out.getBytes();
}

Making this logic generic enough to handle v0, upgrade-from, and v1 will be cumbersome.

Why?

I understand that putting variable length at the end could be simpler. I guess I am just used to how it's usually done if Kafka. And personally, if I read code it's easier to have a mental model of the different versions in the bytes, if older versions are always a prefix of newer versions?

[Gerrrr]

Fixed in 13e772e.

[mjsax]

upgrading from 3.3

Should be 3.2 -- the fix should go into 3.3, right?

[Gerrrr]

Right! Fixed in cd8ba18.

[mjsax]

When upgrading from 2.4 to a newer version it is not required to specify this config.

Seems this needs an update?

[Gerrrr]

Good catch! Fixed in cd8ba18.

[mjsax]

We don't need this check. StreamsConfig ensures that it will be a String. Cf StreamConfig:

.define(UPGRADE_FROM_CONFIG,
        ConfigDef.Type.STRING,

[Gerrrr]

Replaced with a not-null check in 8ebbaaf.

[mjsax]

As above.

[Gerrrr]

Replaced with a not-null check in 8ebbaaf.

[mjsax]

This means that we can't just put data.getVersion() into the buffer, but have to hardcode it to 0

Yes, but this seems ok? Ie, we would have a serializeV0Internal(final short version) that is called as (something like):

public byte[] serializeV0() {
  return serializeV0Internal(0).getBytes();
}

public byte[] serializeV1() {
  final ByteStream out = serializeV0Internal(0);
  
  // add v1 stuff to `out`
  
  return out.getBytes();
}

Making this logic generic enough to handle v0, upgrade-from, and v1 will be cumbersome.

Why?

I understand that putting variable length at the end could be simpler. I guess I am just used to how it's usually done if Kafka. And personally, if I read code it's easier to have a mental model of the different versions in the bytes, if older versions are always a prefix of newer versions?

[Gerrrr]

@mjsax Can you please take another look?

[mjsax]

Overall LGTM -- just wondering if we need to change the subscription-response wire format?

Also (can also be a follow up PR): we should add a system test for the upgrade to make sure it really works) -- we have already upgrade tests, so maybe it's sufficient to add a FK-join to this test topology to cover this case?

[mjsax]

nit:

// version 0x00 fields:

Same below:

// version 0x01 fields:

Or just change to 1 instead of 0x01 (compare Subscription.java -- might be nice to align both).

[Gerrrr]

Fixed in f7f9181#diff-4a79c25b7d01993f00f2ffa554e15f18a2c4f6e1e0f5c18dc33ad24015261f22R24-R29

[mjsax]

While we need this field in this class, so we actually need to serialize it? It seems it must not go over the wire, does it?

Thus, for the subscription response, we don't need the dump the version? -- We only need it on the right hand side to write the response into the topic, but we can do this in-memory only?

If this is correct, we also don't need to update the corresponding Serde.

[Gerrrr]

Good catch! Done in f7f9181.

[mjsax]

nit:

// v0 fields:

Same for v1 below

[Gerrrr]

f7f9181#diff-f9c7a23b1e3f38e1a7c00d05f8d5fa725df7a8147a0d44818d3b63cbf7b0ad23R28

[mjsax]

nit: indention

[Gerrrr]

Fixed in 1361497.

[mjsax]

nit (formatting -- hard to see where method signature end and code starts):

    final int extraLength
) {
    final byte[] serializedData = data.getForeignValue() == null ? null : serializer.serialize(topic, data.getForeignValue());

[mjsax]

nit: remove empty line

[mjsax]

Merged to trunk. Thanks for the fix!