<a id="messagetimestamprouter"></a>

# Kafka Connect MessageTimestampRouter SMT Usage Reference for Confluent Platform

The `MessageTimestampRouter` SMT
(`io.confluent.connect.transforms.MessageTimestampRouter`) updates a record’s
topic name using the original topic value and a timestamp extracted from the
message value.

## Description

Update the record’s topic field as a function of the original topic value and
the record’s timestamp field.

This SMT is useful for sink connectors because the topic field often
determines the equivalent entity name in the destination system (for example,
a database table or search index name). This SMT extracts the timestamp from
the message value’s specified field, which is especially useful for log data
in which the timestamp is stored as a field in the message. The message value
must be a [Map instance](https://docs.confluent.io/platform/current/connect/javadocs/javadoc/org/apache/kafka/connect/data/Schema.Type.html#MAP)
([Structs](https://docs.confluent.io/platform/current/connect/javadocs/javadoc/org/apache/kafka/connect/data/Struct.html) are not supported).
See [Kafka Connect TimestampRouter SMT Usage Reference for Confluent Platform](timestamprouter.md#timestamprouter) to specify a basic topic pattern and timestamp
format.

## Installation

This transformation is developed by Confluent and does not ship by default
with Apache Kafka® or Confluent Platform.
You can install this transformation using the
[confluent connect plugin
install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html)
command:

```none
confluent connect plugin install confluentinc/connect-transforms:latest
```

<a id="messagetimestamprouter-properties"></a>

## Properties

| Name                       | Description                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   | Type   | Default                 | Valid values   | Importance   |
|----------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|-------------------------|----------------|--------------|
| `topic.format`             | Format string which can contain `${topic}` and `${timestamp}` as placeholders for the topic and timestamp, respectively.                                                                                                                                                                                                                                                                                                                                                                      | string | `${topic}-${timestamp}` |                | high         |
| `message.timestamp.format` | Format string for the message’s timestamp that is compatible with `java.time.format.DateTimeFormatter`. For additional details, see [DateTimeFormatter](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html). If no configuration or an empty string is provided, defaults to the format string for timestamp of [ISO8601 standard](https://www.joda.org/joda-time/apidocs/org/joda/time/format/ISODateTimeFormat.html), with mandatory date and optional time. | string | “”                      |                | high         |
| `topic.timestamp.format`   | Format string for the topic’s timestamp that is compatible with `java.time.format.DateTimeFormatter`. For additional details, see [DateTimeFormatter](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html).                                                                                                                                                                                                                                                     | string | yyyy.MM.dd              |                | high         |
| `message.timestamp.keys`   | Comma-separated list of field names to look up the timestamp in the message value, in the order the names are listed. The timestamp is taken from the first found field.                                                                                                                                                                                                                                                                                                                      | list   |                         |                | high         |

## Example

The following example extracts a field named `timestamp`, `time`, or `ts`
from the message value, in the order specified by the `message.timestamp.keys`
configuration. This timestamp value is originally in the format specified by
`message.timestamp.format`. It adds a topic prefix and appends the timestamp
of the format specified by `topic.timestamp.format` to the message topic.

```json
"transforms": "MessageTimestampRouter",
"transforms.MessageTimestampRouter.type": "io.confluent.connect.transforms.MessageTimestampRouter",
"transforms.MessageTimestampRouter.topic.format": "foo-${topic}-${timestamp}",
"transforms.MessageTimestampRouter.message.timestamp.format": "yyyy-MM-dd",
"transforms.MessageTimestampRouter.topic.timestamp.format": "yyyy.MM.dd",
"transforms.MessageTimestampRouter.message.timestamp.keys": "timestamp,time,ts"
```

Message value: `{"time":"2019-08-06"}`

Topic (before): `bar`

Topic (after): `foo-bar-2019.08.06`

## Predicates

Configure transformations with predicates to ensure they process only records
that satisfy a particular condition. You can also use predicates in a
transformation chain with the [Kafka Connect Filter (Kafka) SMT Usage Reference for Confluent Platform](filter-ak.md#ak-filter) to conditionally filter specific
records. For more information, see [Predicates](filter-ak.md#predicates).