Support consumer offset metadata from Kafka source records

[jcrean]

Currently, the consumer code in KafkaCustomConsumer is grabbing the topic/partition/timestamp info from the source ConsumerRecord and adding them as attributes in the event metadata. It would be helpful to also have access to the offset field.

This would allow us to add calls to getMetadata("kafka_offset") in our pipelines, as we use this for internal tracking/auditing. Looking at the code, it seems like it would be relatively easy to add this.

[dlvenable]

@jcrean , Are you looking to get the offset within the partition for each event?

Would you be interested in contributing this change? You can see where we do this for other metadata.

data-prepper/data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java

Lines 458 to 463 in 47cead1

	eventMetadata.setAttribute("kafka_timestamp", receivedTimeStamp);
	eventMetadata.setAttribute("kafka_timestamp_type", consumerRecord.timestampType().toString());
	eventMetadata.setAttribute("kafka_topic", topicName);
	eventMetadata.setAttribute("kafka_partition", String.valueOf(partition));
	eventMetadata.setExternalOriginationTime(Instant.ofEpochMilli(receivedTimeStamp));
	event.getEventHandle().setExternalOriginationTime(Instant.ofEpochMilli(receivedTimeStamp));

[jcrean]

@dlvenable Yes exactly, each ConsumerRecord contains the offset within the partition, which is data that we often use internally for tracking/investigations. I'd be happy to take a stab at a PR. Seems like a fairly straightforward change to make.

[jcrean]

Hey @dlvenable sorry for the delay. Work has been a bit crazy lately. I finally got around to a PR for this:
#5331