fixes #327: Improve error message when Kafka Streams magic number found

[nachomdo]

Fixes: #327

36:31.314 INFO  [pc-broker-poll]  (AbstractParallelEoSStreamProcessor.java:387#onPartitionsAssigned) Assigned 3 total (3 new) partition(s) [input-topic-0, input-topic-1, input-topic-2]
36:31.343 ERROR [pc-broker-poll]  (PartitionStateManager.java:116#onPartitionsAssigned) Error in onPartitionsAssigned
io.confluent.parallelconsumer.offsets.KafkaStreamsV2EncodingNotSupported: Kafka Streams offset metadata V2 not supported
	at io.confluent.parallelconsumer.offsets.EncodedOffsetPair.getDecodedIncompletes(EncodedOffsetPair.java:120) ~[classes/:na]
	at io.confluent.parallelconsumer.offsets.OffsetMapCodecManager.decodeCompressedOffsets(OffsetMapCodecManager.java:246) ~[classes/:na]
	at io.confluent.parallelconsumer.offsets.OffsetMapCodecManager.deserialiseIncompleteOffsetMapFromBase64(OffsetMapCodecManager.java:168) ~[classes/:na]
	at io.confluent.parallelconsumer.offsets.OffsetMapCodecManager.deserialiseIncompleteOffsetMapFromBase64(OffsetMapCodecManager.java:158) ~[classes/:na]
	at io.confluent.parallelconsumer.offsets.OffsetMapCodecManager.decodePartitionState(OffsetMapCodecManager.java:172) ~[classes/:na]
	at io.confluent.parallelconsumer.offsets.OffsetMapCodecManager.lambda$loadPartitionStateForAssignment$0(OffsetMapCodecManager.java:133) ~[classes/:na]
	at java.base/java.util.HashMap.forEach(HashMap.java:1421) ~[na:an]

Checklist

• [] Documentation (if applicable)
• [] Changelog

[astubbs]

Good work!

We only need one exception - if that's the pattern to use
can you include a reference for where you got the magic number info from? byte 1 and 2

if this is the error that's printed:
"Kafka Streams offset metadata V1 not supported"
we need something more friendly. should be something like "it looks like you're reusing a kafka streams consumer group id with PC. This isn't supported - you should use a fresh consumer group, unique to your use of pc" - something like that?

[what-the-diff]

• Added KafkaStreamsEncodingNotSupported exception
• Added support for Kafka Streams magic number in OffsetEncoding enum
• Updated CloseAndOpenOffsetTest to skip new encoding types (KafkaStreams and KafkaStreamsV2) when running tests on offsets open/close cycle
• Updated ensureEncodingGracefullyWorksWhenOffsetsAreVeryLargeAndNotSequential test to skip new encoding types (KafkaStreams and KafkaStreamsV2)
• Update WorkManagerOffsetMapCodecManagerTest with two additional deserialise tests, one for each version of the kafka streams metadata format: v1 & v2

[astubbs]

just about there! can close this one up quick now.

[nachomdo]

@astubbs Thanks for the review! I will apply the suggested feedback today so we can close this 😊