Kafka - Inspecting outgoing kafka messages.

[MintInfusion]

Hi, we are currently trying to automatically propagate an incoming HTTP Header in Quarkus as a kafka header in our microservice architecture for testing purposes. Let's call that header "X-Testing-Data" for now.

My current plan is to use channel decorators to do the following steps:

1. Inspect the outgoing kafka message, if it has already set "X-Testing-Data" as a kafka header, do nothing.
2. If the header is absent, see if "X-testing-data" is found in an HTTP message context and use that if available to populate the message.

Unfortunately, I am currently already failing with step 1. Outgoing messages seem to have varying formats, so to figure out where the Kafka Headers inside the message are, I would have to reimplement a subset of this function here:

smallrye-reactive-messaging/smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/KafkaSink.java

Lines 176 to 249 in 8bfcb7f

	return message -> {
	try {
	Optional<OutgoingKafkaRecordMetadata<?>> om = getOutgoingKafkaRecordMetadata(message);
	OutgoingKafkaRecordMetadata<?> outgoingMetadata = om.orElse(null);
	String actualTopic = outgoingMetadata == null || outgoingMetadata.getTopic() == null ? this.topic
	: outgoingMetadata.getTopic();
	ProducerRecord<?, ?> record;
	OutgoingCloudEventMetadata<?> ceMetadata = message.getMetadata(OutgoingCloudEventMetadata.class)
	.orElse(null);
	IncomingKafkaRecordMetadata<?, ?> incomingMetadata = getIncomingKafkaRecordMetadata(message).orElse(null);
	if (message.getPayload() instanceof ProducerRecord) {
	record = (ProducerRecord<?, ?>) message.getPayload();
	} else if (writeCloudEvents && (ceMetadata != null || mandatoryCloudEventAttributeSet)) {
	// We encode the outbound record as Cloud Events if:
	// - cloud events are enabled -> writeCloudEvents
	// - the incoming message contains Cloud Event metadata (OutgoingCloudEventMetadata -> ceMetadata)
	// - or if the message does not contain this metadata, the type and source are configured on the channel
	if (writeAsBinaryCloudEvent) {
	record = KafkaCloudEventHelper.createBinaryRecord(message, actualTopic, outgoingMetadata,
	incomingMetadata,
	ceMetadata, runtimeConfiguration);
	} else {
	record = KafkaCloudEventHelper
	.createStructuredRecord(message, actualTopic, outgoingMetadata, incomingMetadata, ceMetadata,
	runtimeConfiguration);
	}
	} else {
	record = getProducerRecord(message, outgoingMetadata, incomingMetadata, actualTopic);
	}
	if (isTracingEnabled) {
	KafkaTrace kafkaTrace = new KafkaTrace.Builder()
	.withPartition(record.partition() != null ? record.partition() : -1)
	.withTopic(record.topic())
	.withHeaders(record.headers())
	.withGroupId(client.get(ConsumerConfig.GROUP_ID_CONFIG))
	.withClientId(client.get(ConsumerConfig.CLIENT_ID_CONFIG))
	.build();
	kafkaInstrumenter.traceOutgoing(message, kafkaTrace);
	}
	log.sendingMessageToTopic(message, actualTopic);
	@SuppressWarnings({ "unchecked", "rawtypes" })
	Uni<RecordMetadata> sendUni = client.send((ProducerRecord) record);
	Uni<Void> uni = sendUni.onItem().transformToUni(recordMetadata -> {
	OutgoingMessageMetadata.setResultOnMessage(message, recordMetadata);
	log.successfullyToTopic(message, recordMetadata.topic(), recordMetadata.partition(),
	recordMetadata.offset());
	return Uni.createFrom().completionStage(message.ack());
	});
	if (this.retries == Integer.MAX_VALUE) {
	uni = uni.onFailure(this::isRecoverable).retry()
	.withBackOff(Duration.ofSeconds(1), Duration.ofSeconds(20)).expireIn(deliveryTimeoutMs);
	} else if (this.retries > 0) {
	uni = uni.onFailure(this::isRecoverable).retry()
	.withBackOff(Duration.ofSeconds(1), Duration.ofSeconds(20)).atMost(this.retries);
	}
	return uni
	.onFailure().recoverWithUni(t -> {
	// Log and nack the messages on failure.
	log.nackingMessage(message, actualTopic, t);
	return Uni.createFrom().completionStage(message.nack(t));
	});
	} catch (RuntimeException e) {
	log.unableToSendRecord(e);
	return Uni.createFrom().failure(e);
	}
	};

But checking for the existence of certain classes inside of the message seems very likely to break in future updates. I did also not find anything about this in the API documentation.

Am I missing something? I assume adding/removing/manipulating key/value/header is the goal of channel decorators, but without a way to inspect the existing message, they seem unusable to me...

What I would have expected to see is a function that transforms any given message into a ProducerRecord or similar.

Hope someone can explain what I am failing to understand.

[ozangunalp]

If you want to propagate HTTP headers the best option is to use OutgoingInterceptor and add OutgoingKafkaRecordMetadata to your outgoing message.

In Quarkus we recommend using ContextLocals to propagate contextual information between HTTP and RM producer channels.

You can write a ServerRequestFilter to inject headers to be propagated to the ContextLocals and retrieve them in your OutgoingInterceptor. In normal use cases, the context is preserved.

I'll write a section with examples about this in the Quarkus Kafka guide.