是否存在可插入时间戳字段的Kafka Connect转换？

Absolutely! You can easily achieve this using Kafka Connect's built-in InsertField transformation—it's designed exactly for adding metadata fields (including timestamps) to your records without writing custom code.

1. Using the Built-in InsertField Transformation #

This is the simplest and most straightforward approach for your use case. Here's how to configure it to add a Unix timestamp field matching your example:

Sample Connector Configuration

{
  "name": "your-target-connector",
  "config": {
    // ... keep your existing connector configurations here ...
    "transforms": "addTimestamp",
    "transforms.addTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    // Define the field name you want to add (matches your "foo" example)
    "transforms.addTimestamp.processing.time.field": "foo",
    // Output Unix timestamp in seconds (use "unix_ms" if you need milliseconds)
    "transforms.addTimestamp.timestamp.format": "unix"
  }
}

Key Details:

• Processing Time vs. Record Timestamp:
  • Use processing.time.field to add the timestamp when Kafka Connect processes the record.
  • If you want to use the original message's timestamp (from Kafka's record metadata, e.g., when the message was produced), replace processing.time.field with timestamp.field.
• Timestamp Format:
  • unix gives you a numeric timestamp in seconds since the epoch (like your example 1589889270).
  • unix_ms gives milliseconds instead.
  • You can also use a custom date format string (e.g., yyyy-MM-dd HH:mm:ss) if you need a human-readable timestamp instead of a numeric value.

2. Custom Transformation (For Advanced Scenarios) #

If you need more control—like custom time zone handling, deriving timestamps from existing fields, or complex formatting—you can build a custom Kafka Connect transformation. But for your specific requirement of adding a simple Unix timestamp field, the built-in InsertField is more than enough.

Example Result #

Input record:

{ "first": "second" }

Transformed output record:

{ "first": "second", "foo": 1589889270 }

内容的提问来源于stack exchange，提问作者abergmeier