Elasticsearch sink always using Kafka timestamp as @timestamp

[xavirg]

A note for the community

• Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
• If you are interested in working on this issue or have submitted a pull request, please leave a comment

Problem

When using the Elasticsearch sink data_stream mode, Vector has a remap method to create the ELK @timestamp field before pushing the event:

vector/src/sinks/elasticsearch/sink.rs

Lines 118 to 121 in 7cfc9c5

	if let Some(cfg) = mode.as_data_stream_config() {
	cfg.sync_fields(&mut log);
	cfg.remap_timestamp(&mut log);
	};

vector/src/sinks/elasticsearch/config.rs

Lines 419 to 428 in 7cfc9c5

	/// If there is a `timestamp` field, rename it to the expected `@timestamp` for Elastic Common Schema.
	pub fn remap_timestamp(&self, log: &mut LogEvent) {
	if let Some(timestamp_key) = log.timestamp_path().cloned() {
	if timestamp_key.to_string() == DATA_STREAM_TIMESTAMP_KEY {
	return;
	}
	log.rename_key(&timestamp_key, event_path!(DATA_STREAM_TIMESTAMP_KEY));
	}
	}

However, I've found that @timestamp value is always the Kafka message time (i.e. %kafka.timestamp), even if the global timestamp field is set to anything else:

log_schema:
  timestamp_key: "custom_time"

Which I would expect to change timestamp_key = "custom_time" at remap_timestamp method, so its value is used for the new @timestamp.

Right now, to bypass this situation, I'm manually overwriting %kafka.timestamp in a transform:

transforms:
  timestamp_fix:
    type: "remap"
    inputs:
      - "kafka"
    source: |-
      %kafka.timestamp = .custom_time

Configuration

acknowledgements:
  enabled: true
schema:
  log_namespace: true
log_schema:
  timestamp_key: "custom_time"

sources:
  kafka:
    type: kafka
    bootstrap_servers: "kafka:9092"
    topics:
    - logging.service.v2.ingestion
    session_timeout_ms: 10000
    drain_timeout_ms: 5000
    commit_interval_ms: 5000
    librdkafka_options:
      partition.assignment.strategy: "cooperative-sticky"
      enable.auto.commit: "true"
      heartbeat.interval.ms: "3000"
    group_id: "test.groupid"
    tls:
      enabled: false
    decoding:
      codec: "json"

sinks:
  opensearch:
    type: elasticsearch
    api_version: v8
    inputs:
      - kafka
    endpoints:
      - "https://opensearch:9200/"
    auth:
      strategy: "basic"
      user: "foo"
      password: "bar"
    mode: data_stream
    bulk:
      action: "create"
    data_stream:
      type: "service"
      dataset: "v2"
      namespace: ""
    batch:
      max_bytes: 10000000
    request:
      timeout_secs: 60
    tls:
      verify_certificate: false
      verify_hostname: false

Version

0.46

Debug Output

Example Data

Event ingested at the Kafka topic:

{
	"Value": "{\"custom_time\":\"2025-04-14T07:19:24.649687061Z\",\"log\":\"Failed to fetch the Moon\\n\"}",
	"Offset": 1,
	"Key": null,
	"Partition": 1,
	"Headers": {},
	"Timestamp": "2025-04-15T09:52:34.778Z"
}

Vector's console output:

{"custom_time":"2025-04-14T07:19:24.649687061Z","log":"Failed to fetch the Moon\n"}

Resulting Elasticsearch document:

{
  "_index": "service-v2",
  "_id": "YyzdOJYB-GTU1tpUZXPf",
  "_score": 1,
  "_source": {
    "@timestamp": "2025-04-15T09:52:34.778Z",
    "custom_time": "2025-04-14T07:19:24.649687061Z",
    "data_stream": {
      "dataset": "v2",
      "namespace": "",
      "type": "service"
    },
    "log": "Failed to fetch the Moon\n"
  },
  "fields": {
    "@timestamp": [
      "2025-04-15T09:52:34.778Z"
    ]
  }
}

Additional Context

No response

References

No response

[pront]

Hi @xavirg,

I noticed you set log_namespace: true which changes how Vector accesses the timestamp field.

Basically the following part of the config, is only relevant when namespacing is OFF (default):

log_schema:
  timestamp_key: "custom_time"

Instead you have to use set_semantic_meaning:

set_semantic_meaning(.custom_time, "timestamp")

---

Implementation here:

    /// Fetches the "timestamp" path of the event. This is either from the "timestamp" semantic meaning (Vector namespace)
    /// or from the timestamp key set on the "Global Log Schema" (Legacy namespace).
    pub fn timestamp_path(&self) -> Option<&OwnedTargetPath> {
        match self.namespace() {
            LogNamespace::Vector => self.find_key_by_meaning("timestamp"),
            LogNamespace::Legacy => log_schema().timestamp_key_target_path(),
        }
    }

This comes up every now and then. There are more details in this issue: #22341 (comment)

[xavirg]

@pront

I tried with log namespacing disabled and still does not work as expected (Kafka timestamp is sent to Elasticsearch sink).

Can you please try to reproduce it at least?

Maybe you closed this issue too fast.