...

Current state: "Under Discussion"

Discussion thread: here

JIRA: here

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Draft implementation: https://github.com/apache/kafka/pull/12647

Motivation

Currently, Sink and Source Connectors include only putBatchLatency metric to measure latency metrics covering only the time expend interacting with the external systems —

• put-batch-latency  metric measures the time to sink a batch of records, and
• poll-batch-time metric measures the time to poll a batch of records from the external system.

At the moment it's difficult to understand another latency aspects, e.g.:

• how much latency has the connector introduced in the process, or
• how long after being written a record is processed.

In order to observe the connector's performance and measure its into a external system.In other to correlate these metrics and measure a complete end-to-end latency of the Sink Connector there is a need for additional measures:from sources to sinks, this KIP is proposing the following additional measurements:

• In the source connector:
  • After polling, there are transformations and conversions that happen before the records are sent to Kafka.
• In the sink connector:
  • Record
  record e2e
  • latency: wall-clock time - record timestamp  to evaluate how late records are processed.
  convert time: latency to convert + transform batch of records to know how long is taking to convert and transform records

  • Convert and transform time before sending records to a external system

With these enhanced metrics , it will be clearer how much latency the sink connector.

In the case of source connectors, convert time can be added to improve the latency monitoring and have parity with the metrics on the sink connectors.

Public Interfaces

The following metrics would be added:

• sink-record-e2e-latency-min [ms]

• sink-record-e2e-latency-max [ms]

• sink-record-e2e-latency-avg [ms]

• sink-record-convert-time-max [ms]

• sink-record-convert-time-avg [ms]

• source-record-convert-time-max [ms]

• source-record-convert-time-avg [ms]

Proposed Changes

Sink Connectors have 3 stages:

available, operators/developers could:

• Monitor and alert when sink processing is happening after accepted latency (e.g. > 5secs)
  • Current workaround:
    • Inserting records timestamp into the target system to handle the calculations there.
• Observe processing lifecycle and monitor on spikes caused by convert and transforms and reduce the time to remediation
  • Current workaround:
    • Infer connector's latency by capturing the timestamp of the source system and comparing it with the Kafka record timestamp; which is not trivial.
    • With the current poll/put and the produce/fetch latencies are not enough to infer the time taken in between.
• Performance testing transforms in non-production environments to measure how latency increases related to throughput.

Proposed Changes

                 ┌───────────────────────────────────────────────────────────────────────────────────────────┐
                 │                                   Source Connector Task                                   │
                 │ ┌──────────────────┐  ┌────────────────┐        ┌─────────┐         ┌─────────────────┐   │
                 │ │ Connector        ├─►│ TransformChain ├───────►│ Convert ├────────►│     Producer    │   │
                 │ │ Implementation   │  └────────────────┘        └─────────┘         │                 │   │  ┌───────┐
┌────────────┐   │ └──────────────────┘  *[tx.chain-lat] INFO  *[convert-latency] INFO └─────────────────┘   ├─►│ Kafka │
│ Ext.System ├──►│  [poll-source-batch]   ┌────┬────┬───┬────┐   ┌────────────────────┐ [request-latency]    │  └───────┘
└────────────┘   │                        │ Tx1│ Tx2│...│TxN │   │Conv.Key|Val|Headers│                      │
                 │                        └────┴────┴───┴────┘   └────────────────────┘                      │
                 │                        *[tx.N-lat] DEBUG    *[convert-X-latency]DEBUG                     │
                 │ (=======batch=======)=(=========================per-record============================)   │
                 └───────────────────────────────────────────────────────────────────────────────────────────┘


                 ┌───────────────────────────────────────────────────────────────────────────────────────────┐
                 │                                     Sink Connector Task                                   │
                 │ ┌──────────────────┐  ┌────────────────┐        ┌─────────┐         ┌─────────────────┐   │
                 │ │ Connector        │◄─┤ TransformChain │◄───────┤ Convert │◄────────┤     Consumer    │   │
                 │ │ Implementation   │  └────────────────┘        └─────────┘         │                 │   │  ┌───────┐
┌────────────┐   │ └──────────────────┘  *[tx.chain-lat] INFO  *[convert-latency] INFO └─────────────────┘   │◄─┤ Kafka │
│ Ext.System │◄──┤  [put-batch-latency]   ┌────┬────┬───┬────┐   ┌────────────────────┐  [fetch-latency]     │  └───────┘
└────────────┘   │                        │ Tx1│ Tx2│...│TxN │   │Conv.Key|Val|Headers│                      │
                 │                        └────┴────┴───┴────┘   └────────────────────┘                      │
                 │                        *[tx.N-lat] DEBUG    *[convert-X-latency]DEBUG                     │
                 │ (=====batch=======)===(===========per-record=======================)==(=====batch=====)   │
                 └───────────────────────────────────────────────────────────────────────────────────────────┘

(*) New metrics

Source Connectors have the following stages:

• task implementation polling records: gather batch of records from external system

• transform: apply transformation chain individually
• convert: convert records to ProducerRecord individually
• send records: send records to Kafka topics individually

The stage for polling records already has a latency metric: poll-batch-time .

transform-chain-source-record-time and convert-source-record-time metric will measure the transformations applied and conversion from SourceRecord  into ProducerRecord .

The stage for sending records can be monitored with Producer sender metrics, e.g. request-latency-avg/max

Sink Connectors have the following stages:

• consumer polling record: gather batch of consumer records

• convert: convert records individually to generic SinkRecord ,

• transform: apply transformation chain

• polling: gather consumer records

• convert/transform: convert records individually to generic SinkRecord  and apply transformers

• process: put record batches into a external system.

Process The processing stage already has a latency metric: put-batch-time 

To measure sink's record e2e latency sink-record-e2e-latency latency (i.e. processing time - event time), it's proposed to measure the different difference between record timestamp and current system time (wall-clock) at the beginning of just before the convert stage as it is when records are iterated already.

Convert latency convert-sink-record-converttime and transform-chain-sink-record-time  measures the convert and transformation per-record.

Polling can be monitored with Consumer fetch metrics, e.g. fetch-latency-avg/max

Public Interfaces

The following metrics would be added at the Task Level:

kafka.connect:type=sink-task-metrics,connector="{connector}",task="{task}"

...

kafka.connect:type=source-task-metrics,connector="{connector}",task="{task}"

kafka.connect:type=sink-task-transform-metrics,connector="{connector}",transform="{transform_alias}",task="{task}"

kafka.connect:type=source-task-transform-metrics,connector="{connector}",transform="{transform_alias}",task="{task}"

Where alias  is the Transform alias name used in configuration:

"transforms": "routeRecords",
"transforms.routeRecords.type": "org.apache.kafka.connect.transforms.RegexRouter"

"routeRecords" in this example.

More granular metrics are recorded at DEBUG level to avoid performance impact.

TransformationChain  and ConnectorConfig  will change their following APIs to support keeping transform alias to record metrics:

ConnectorConfig:

-    public <R extends ConnectRecord<R>> List<Transformation<R>> transformations() {
+    public <R extends ConnectRecord<R>> LinkedHashMap<String, Transformation<R>> transformations() {

TransformChain:

-    public TransformationChain(List<Transformation<R>> transformations, RetryWithToleranceOperator retryWithToleranceOperator) {
+    public TransformationChain(LinkedHashMap<String, Transformation<R>> transformations, RetryWithToleranceOperator retryWithToleranceOperator) {

Compatibility, Deprecation, and Migration Plan

ConnectConfig and TransformationChain users will have to migrate to the new interfaces. Though these APIs are used internally on Worker instantiations of Tasks and not meant for external usage.

On the other hand, in the Source Connector has:

• polling: gather records from external system

• transform/convert: apply transformations and convert records to ProducerRecord 
• send records: to Kafka topics

Polling records stage already has a latency metric: poll-batch-time .

source-record-convert-time metric will measure the transformations applied and conversion from SourceRecord  into ProducerRecord .

Send records stage can be monitored with Producer sender metrics, e.g. request-latency-avg/max 

Compatibility, Deprecation, and Migration Plan

N/A

Rejected Alternatives

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.

...