feat: allow the Kafka record key deserializer to be configured

[javsanbel2]

📝 Description

KafkaMessageReaderBuilder hardcodes a LongDeserializer for the record key, matching the key written by the Apiary Hive Metastore listener. That is a reasonable default, but it currently cannot be changed — properties passed to withConsumerProperties() are merged with collisions resolved in favour of the builder's own defaults:

consumerProperties.forEach((key, value) -> props.merge(key, value, (v1, v2) -> v1));

A caller setting key.deserializer there is silently ignored, despite the README stating that additional consumer properties are configurable.

Reading a topic whose producer keys records with another type (for example a String key) then fails with:

SerializationException: Size of data received by LongDeserializer is not 8

This is worse than one rejected record. The consumer position does not advance past a record it cannot deserialize, so the reader retries the same offset indefinitely and makes no progress. Because the exception surfaces from poll(), a caller that catches per-event exceptions and continues will spin rather than skip.

Changes

• Add KafkaMessageReaderBuilder.withKeyDeserializer(String) so callers can supply the deserializer matching their producer.
• The default remains LongDeserializer, and the precedence rule for withConsumerProperties() is unchanged, so existing callers are unaffected. Flipping the merge so caller properties win would be a wider behavioural change for every existing caller; an explicit builder method is opt-in and reviewable.
• The consumer key type becomes Object rather than Long. The record key is never used to decode the event — events are read from the record value alone — so this keeps the declared type honest when a different key deserializer is configured.
• Replace the "key.deserializer" / "value.deserializer" string literals with the ConsumerConfig constants, and the deserializer class-name literals with Class#getName().
• Extract property construction into a package-private buildConsumerProperties(), so the resulting configuration can be asserted without standing up a real KafkaConsumer.
• Document the new method and the property-precedence rule in the kafka-metastore-receiver README; add a 8.2.5 entry to CHANGELOG.md (the parent pom is already at 8.2.5-SNAPSHOT).

✅ Test plan

• Six new unit tests in KafkaMessageReaderTest: default deserializer, the override, withConsumerProperties still not winning, non-colliding consumer properties preserved, and null/empty validation.
• mvn test on kafka-metastore-receiver — Tests run: 18, Failures: 0, Errors: 0.
• mvn test on kafka-metastore-integration-tests — Tests run: 7, Failures: 0, Errors: 0, confirming the end-to-end listener/receiver round trip still passes with the default deserializer.
• mvn package from the reactor root — BUILD SUCCESS.
• Reviewer confirmation that no downstream consumer relies on the package-private KafkaMessageReader constructor's KafkaConsumer<Long, byte[]> signature.

🔗 Related Issues

None.

🤖 Generated with opencode