...

{BYTE_ARRAY oldValue}{BYTE newOldFlag=0}
{BYTE_ARRAY newValue}{BYTE newOldFlag=1}
{INTUINT32 newDataLength}{BYTE_ARRAY newValue}{BYTE_ARRAY oldValue}{BYTE newOldFlag=2}

...

Users should implement `.equals` method for key

Another important part of the proposed logic is detecting if the key has changed. In order for us to be able to do this, we depend on a correct implementation of the .equals  method on the key, which users will need to implement. 

Note however that this is not a strict new requirement for users; if users fail to implement the .equals  method method, then they should still get generally get the old behaviour of sending the oldValue and the newValue as two separate messages to the repartition topic i.e. nothing breaks. In this way, our changes can be made considered backwards compatible with existing code where the key type does not implement the .equals  method   method. There is one edge-case where this does not hold true, as follows. 

Since the default .equals implementation for an Object  is by reference, if a user's groupBy returns the same reference for the key, then the oldKey and the newKey will naturally .equals  each other. This will result in a single event being sent to the repartition topic. This change in behaviour should be considered a "bug-fix" rather than a "breaking change" as the semantics of the operation remain unchanged, the only thing that changes for users is they no longer see transient "inconsistent" states.  In the worst case, users in this situation will need to update any strict tests that check specifically for the presence of transient "inconsistent" states. 

Compatibility, Deprecation, and Migration Plan

...