feat(kafka-clients): handle readonly record headers on context injection through producer wrapper

[wpessers]

After #17231 when looking at the code again I noticed some additional exception handling in the interception flow. Talking about the interception flow that has the try/catch around propagator.inject() at:

opentelemetry-java-instrumentation/instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/internal/KafkaProducerTelemetry.java

Lines 64 to 71 in bd6e305

	if (producerPropagationEnabled) {
	try {
	propagator.inject(context, record.headers(), SETTER);
	} catch (Throwable t) {
	// it can happen if headers are read only (when record is sent second time)
	logger.log(WARNING, "failed to inject span context. sending record second time?", t);
	}
	}

I found that, as the comment and log statement there allude to, this is for when a record's headers have already been marked readonly (which happens after serialization through KafkaProducer.doSend()). So this try/catch should be there, in case of retries for example, to gracefully handle the IllegalStateException that is thrown when trying to inject headers into a readonly RecordHeaders instance.

Mea culpa for missing this on the previous PR.

[laurit]

alternatively could use a mock, but this is fine too

[wpessers]

I guess it may be more readable with a mock, I just went with this approach because it mirrors the setup of the existing tests in this same file. Your call though, I'd happily change it to use a mock if preferred!

[wpessers]

@laurit while working on implementing your suggestion I dove a bit deeper into some other parts and realized some error handling we might want to add.

For the producer wrapper, at:

opentelemetry-java-instrumentation/instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/internal/KafkaProducerTelemetry.java

Lines 94 to 101 in bd6e305

	Context context = producerInstrumenter.start(parentContext, request);
	if (producerPropagationEnabled) {
	propagator.inject(context, record.headers(), SETTER);
	}
	try (Scope ignored = context.makeCurrent()) {
	return sendFn.apply(record, new ProducerCallback(callback, parentContext, context, request));
	}

producerInstrumenter.start() starts a span. That span will get ended in ProducerCallback.onCompletion(). If sendFn.apply() throws synchronously (which is well documented that it can happen at https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java#L1052-L1056), the callback is never invoked, meaning the span is never ended.

So I'm thinking we may want to add some additional error handling to explicitly end the span in case of failure so that the span is always properly ended. WDYT?