Feature Request - Specific Kafka Offset in telegraf Kafka Consumer

[biker73]

Feature Request

Proposal:

Currently the Kafka consumer plugin allows you to specify Oldest or Newest as offset values.
It would be very useful to be abler to specify a starting offset number.

Current behavior:

Oldest or Newest specified only work as expected.

Desired behavior:

To specify specify start offset

Use case: [Why is this important (helps with prioritizing requests)]

Kafka is used as a persistent message store to which we send 10's of millions of messages a day. We use telegraf as a consumer to read messages and write them to a number of targets (one being influx).

If any target is unavailable (crashed / maintenance / power failure ) telegraf will drop messages once its buffer is full (which takes a few seconds), and future messages are read from kafka but dropped.

One workaround is to have an additional 'recovery' telegraf agent that re-processes all the messages from kafka. This is started once the target systems that have been unavailable have been brought back on line and can start to accept messages again. Once the backlog has been processed this instance of telegraf is then stopped until it happens again. Each time this is used the recovery process will start from the last offset recorded from the previous use which can be 100's of millions of messages ago or many days / weeks in the past.

It would be useful to be able to specify a starting offset so I only have to re-process messages from the last known good time / offset and not reprocess messages that are known to have been successfully sent to the target system.

[sparrc]

this could also be solved by solving #2240?

[seuf]

Hi,
I don't think it's possible to specify the offset in the kafka consumer input plugin.
the plugin is based on the sarama Go kafka client, and according to it's documentation, only OffsetNewest or OffsetOldest are allowed.

        // Offsets specifies configuration for how and when to commit consumed
        // offsets. This currently requires the manual use of an OffsetManager
        // but will eventually be automated.
        Offsets struct {
            // How frequently to commit updated offsets. Defaults to 1s.
            CommitInterval time.Duration

            // The initial offset to use if no offset was previously committed.
            // Should be OffsetNewest or OffsetOldest. Defaults to OffsetNewest.
            Initial int64

            // The retention duration for committed offsets. If zero, disabled
            // (in which case the `offsets.retention.minutes` option on the
            // broker will be used).  Kafka only supports precision up to
            // milliseconds; nanoseconds will be truncated. Requires Kafka
            // broker version 0.9.0 or later.
            // (default is 0: disabled).
            Retention time.Duration
        }

[sspaink]

As the previous comment pointed out, this is a limitation from the package the plugin is using. Looking at the latest version it doesn't seem this has changed. Closing because there hasn't been any activity in a long time, but if there is someone still interested in feature please comment and we can see what alternatives there might be. Might require an upstream change.