Spring-cloud-contract: Adding kafka message key to a contract, causes assertion fail on producer side

[maciekh]

Consumer side application requires events with filled in kafka message key.
To add a kafka message key to the contract I need to add:

header('kafka_messageKey', 'keyValue')

When I do it, a new assertion is add to the generated contract test:

response.getHeader('kafka_messageKey')?.toString() == 'keyValue'

And of course test fails because no such header is present.

Expected behaviour:
Assertion in generated tests should not be added for 'kafka_messageKey' header.
Or kafka message key should be added in another way.
Eg.:

outputMessage {
    sentTo('...')
    key('keyValue')
    body(...)
    headers {...}
}

[marcingrzejszczak]

I'm no expert in Kafka for that matter, are there any other special headers that we should take into consideration?

[maciekh]

Me neither ;)
I hope there is a problem with message key only.
Kafka message may contains message key, message value and headers.
The problem is that in SCC message key is set via a header field but is not a part of header in kafka message.

[marcingrzejszczak]

Pinging @sobychacko :grimacing:
/cc @garyrussell

[garyrussell]

@maciekh Can you provide more context - including the code that is sending the message and your test case.

kafka_messageKey is a spring-messaging header that gets mapped to the ProducerRecord.key field. On the inbound side, it appears in the ConsumerRecord.key field and is mapped to the spring-messaging kafka_messageKey header.

[garyrussell]

It looks to me that sleuth is dealing with the raw Kafka ProducerRecord and ConsumerRecord and not the spring messaging abstraction so you should be looking for the key property.

https://github.com/spring-cloud/spring-cloud-sleuth/blob/d1dddbf2c9d736f1dc792aa5b8a892e921442747/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceMessagingAutoConfiguration.java#L395-L403

[marcingrzejszczak]

we're in the Spring Cloud Contract project :P Maybe we should check this out https://github.com/spring-cloud/spring-cloud-contract/blob/master/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/KafkaStubMessages.java ?

[garyrussell]

Looks like the problem is here

https://github.com/spring-cloud/spring-cloud-contract/blob/8c20fd3ae8cc2e7fc67a5a9cf64f133627ffc741/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/KafkaStubMessages.java#L139-L161

You are not synthesizing the other headers from the other record fields.

This is the code we use in Spring:

        commonHeaders(acknowledgment, consumer, rawHeaders, record.key(), record.topic(), record.partition(),
                record.offset(), ttName, record.timestamp());
...
    default void commonHeaders(Acknowledgment acknowledgment, Consumer<?, ?> consumer, Map<String, Object> rawHeaders,
            Object theKey, Object topic, Object partition, Object offset,
            @Nullable Object timestampType, Object timestamp) {

        rawHeaders.put(KafkaHeaders.RECEIVED_MESSAGE_KEY, theKey);
        rawHeaders.put(KafkaHeaders.RECEIVED_TOPIC, topic);
        rawHeaders.put(KafkaHeaders.RECEIVED_PARTITION_ID, partition);
        rawHeaders.put(KafkaHeaders.OFFSET, offset);
        rawHeaders.put(KafkaHeaders.TIMESTAMP_TYPE, timestampType);
        rawHeaders.put(KafkaHeaders.RECEIVED_TIMESTAMP, timestamp);
        JavaUtils.INSTANCE
            .acceptIfNotNull(KafkaHeaders.GROUP_ID, MessageConverter.getGroupId(),
                    (key, val) -> rawHeaders.put(key, val))
            .acceptIfNotNull(KafkaHeaders.ACKNOWLEDGMENT, acknowledgment, (key, val) -> rawHeaders.put(key, val))
            .acceptIfNotNull(KafkaHeaders.CONSUMER, consumer, (key, val) -> rawHeaders.put(key, val));
    }

after mapping the ConsumerRecord.headers().

[maciekh]

To simply reproduce this issue you can use spring-cloud-samples code.

https://github.com/spring-cloud-samples/spring-cloud-contract-samples/blob/c44ff80eb3985107edcb106497a0d85b304df1a8/producer_kafka/src/test/resources/contracts/shouldSendFoo.groovy#L3-L13

In this place header

headers {
    header('kafka_messageKey', 'kafka_key')
}

should be added.

After that such assertions appear in generated test:

assertThat(response.getHeader("kafka_messageKey")).isNotNull();
assertThat(response.getHeader("kafka_messageKey").toString()).isEqualTo("kafka_key");

And test failes.

[nekator]

Facing the same problem while creating contracts for my application. Sending the key works but receiving a key does not work.
I debugged KafkaStubMessages.java and saw that the key ist not used while transforming the ConsumerRecord into a Message.

[marcingrzejszczak]

Regardless of this being a bug that should be fixed would the option to use the actual Kafka broker via test containers like here producer and here consumer be ok to you as a workaround? Would it solve your problem?

[marcingrzejszczak]

@garyrussell can you elaborate more on what exactly the problem is and how to solve it?

[garyrussell]

@marcingrzejszczak ConsumerRecord has 3 (primary) properties - key(), value(), and headers() as well as others (timestamp etc).

In Spring messaging, we map the value to the message payload (as do you), the key and other properties get added as headers (e.g. KafkaHeaders.MESSAGE_KEY) in addition to the actual record.headers().

It appears that the stub message is discarding the key and other properties instead of mapping them.

[bono007]

@marcingrzejszczak I am taking a look at this and should have something in for review w/in next 24 hours.

[bono007]

I have fixed the main issue in the ticket and am now synthesizing the other header values back into the consumer record.

However, that revealed another issue in that the Receiver consumer is hardcoded to use <int, String> for its <k, v> SerDe (as per KafkaTestUtils.consumerProps). This revealed itself by the headerValue (in the ticket description) being deserialized into something like 4354456 (its int deserialization form).

I have changed this to leverage the spring.kafka.consumer props. This now lets us use any arbitrary key, value (including json object keys). I am working through one last thing then I will get some tests in place (we are a bit thin wrt to the Kafka messaging ones). Should be wrapped up w/in the next 24 hours.

[bono007]

I should have the submitted tomorrow. I have been fighting a Kafka Streams dumpster fire for the last 24 hours. Just wanted to follow up as my last message said "blah blah blah w/in 24 hours".