Skipping offsets warning in Druid and Kafka transactional messaging

[Erwin Bolwidt]

I'm running Druid 0.11 and I'm getting loads of warnings in the Kafka indexer peon task log for my datasource of the form:

WARN [task-runner-0-priority-0] io.druid.indexing.kafka.KafkaIndexTask - Skipped to offset[16,521,675] after offset[16,521,674] in partition[0].

I'm using Kafka-Streams in transactional mode to transform data before ingesting in Druid.

I have set "skipOffsetGaps": true in my ioConfig, so the ingestion is working correctly. But it seems that Druid doesn't understand that Kafka uses up offsets for commits when using transactions - it's certainly doesn't seem something that should be logged at WARN level as it's expected that Kafka skips offsets with transactions.

I am worried that Druid may have other issues with Kafka Transactions. Is ingestion from topics that have been produced with Kafka transactions supported by Druid or should I expect other issues when doing so?

[Gian Merlino]

Hi Erwin,

The 'skipped offset' check in Druid's Kafka indexing consumer is a sanity check meant to make sure we process every message in order. It was originally put there just in case there were bugs in the Kafka broker, consumer, etc, that caused them to skip messages. This doesn't hold for compacted topics of course, but that's ok since it's not expected that people will want to read from compacted topics into Druid.

I'm not familiar with how Kafka transactions works but it sounds like they break this assumption too. That's more interesting since I do expect people to want to read from these topics into Druid. Do you have some pointers to how this works, especially on the consumer side? It sounds like we should change the Druid code a bit.

Gian

--
You received this message because you are subscribed to the Google Groups "Druid User" group.
To unsubscribe from this group and stop receiving emails from it, send an email to druid-user+unsubscribe@googlegroups.com.
To post to this group, send email to druid...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/druid-user/f6e9c41d-cd4a-4e82-901c-2ab5f5d6c117%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Erwin Bolwidt]

Hi Gian,

The following Confluent blog is a good overview:

https://www.confluent.io/blog/transactions-apache-kafka/

In my application, I enable the use of transactions by turning on "exactly once" semantics in Kafka Streams:

https://www.confluent.io/blog/enabling-exactly-kafka-streams/

Best Regards

Erwin Bolwidt

On Wednesday, 14 February 2018 14:43:41 UTC+8, Gian Merlino wrote:

Hi Erwin,

The 'skipped offset' check in Druid's Kafka indexing consumer is a sanity check meant to make sure we process every message in order. It was originally put there just in case there were bugs in the Kafka broker, consumer, etc, that caused them to skip messages. This doesn't hold for compacted topics of course, but that's ok since it's not expected that people will want to read from compacted topics into Druid.

I'm not familiar with how Kafka transactions works but it sounds like they break this assumption too. That's more interesting since I do expect people to want to read from these topics into Druid. Do you have some pointers to how this works, especially on the consumer side? It sounds like we should change the Druid code a bit.

Gian

On Tue, Feb 13, 2018 at 10:38 PM, Erwin Bolwidt <ebol...@gmail.com> wrote:

I'm running Druid 0.11 and I'm getting loads of warnings in the Kafka indexer peon task log for my datasource of the form:

WARN [task-runner-0-priority-0] io.druid.indexing.kafka.KafkaIndexTask - Skipped to offset[16,521,675] after offset[16,521,674] in partition[0].

I'm using Kafka-Streams in transactional mode to transform data before ingesting in Druid.

I have set "skipOffsetGaps": true in my ioConfig, so the ingestion is working correctly. But it seems that Druid doesn't understand that Kafka uses up offsets for commits when using transactions - it's certainly doesn't seem something that should be logged at WARN level as it's expected that Kafka skips offsets with transactions.

I am worried that Druid may have other issues with Kafka Transactions. Is ingestion from topics that have been produced with Kafka transactions supported by Druid or should I expect other issues when doing so?

--
You received this message because you are subscribed to the Google Groups "Druid User" group.

To unsubscribe from this group and stop receiving emails from it, send an email to druid-user+...@googlegroups.com.

[Erwin Bolwidt]

Hi Gian,

I do think there are additional issues, but I'm not sure yet if there are inherent in the kafka 0.10.2.0 client libraries or in the way that druid uses them.

I've had the situation twice now in which the kafka indexing consumer was stuck without any messages in the log files (not even the "offset skipped" warning that I normally get a lot). In both situations, the currentOffsets in the "/druid/indexer/v1/supervisor/[name]/status" endpoint was pointing at a non-existing offset (a commit message).

With a small Java client (which initially has some classpath issues), I see that the problem is likely in the 0.10.2.0 client, which stops reading at a commit message - it keeps returning zero ConsumerRecords. The "skipOffsetGaps" feature normally manages to skip over, but likely that doesn't work if the KafaIndexerTask happens to start exactly at an offset that points to a commit record. I guess when Druid upgrades the kafka client to 0.11 or higher, the problem will go away by itself.

I've had to turn on the "exactly once" guarantee in my Kafka Streams transformation application, which is not ideal. But hopefully that will make the hangs go away.

(Note: it would be nice to have a REST endpoint to explicitly set the offsets for each partition for the indexer, rather than only a general "reset" endpoint. That would be useful for debugging in other situations too)

Best

Erwin

On Wednesday, 14 February 2018 14:43:41 UTC+8, Gian Merlino wrote:

Hi Erwin,

The 'skipped offset' check in Druid's Kafka indexing consumer is a sanity check meant to make sure we process every message in order. It was originally put there just in case there were bugs in the Kafka broker, consumer, etc, that caused them to skip messages. This doesn't hold for compacted topics of course, but that's ok since it's not expected that people will want to read from compacted topics into Druid.

I'm not familiar with how Kafka transactions works but it sounds like they break this assumption too. That's more interesting since I do expect people to want to read from these topics into Druid. Do you have some pointers to how this works, especially on the consumer side? It sounds like we should change the Druid code a bit.

Gian

On Tue, Feb 13, 2018 at 10:38 PM, Erwin Bolwidt <ebol...@gmail.com> wrote:

I'm running Druid 0.11 and I'm getting loads of warnings in the Kafka indexer peon task log for my datasource of the form:

WARN [task-runner-0-priority-0] io.druid.indexing.kafka.KafkaIndexTask - Skipped to offset[16,521,675] after offset[16,521,674] in partition[0].

I'm using Kafka-Streams in transactional mode to transform data before ingesting in Druid.

I have set "skipOffsetGaps": true in my ioConfig, so the ingestion is working correctly. But it seems that Druid doesn't understand that Kafka uses up offsets for commits when using transactions - it's certainly doesn't seem something that should be logged at WARN level as it's expected that Kafka skips offsets with transactions.

I am worried that Druid may have other issues with Kafka Transactions. Is ingestion from topics that have been produced with Kafka transactions supported by Druid or should I expect other issues when doing so?

--
You received this message because you are subscribed to the Google Groups "Druid User" group.

To unsubscribe from this group and stop receiving emails from it, send an email to druid-user+...@googlegroups.com.

[Gian Merlino]

Thanks for the research Erwin, I have filed this Druid issue: https://github.com/druid-io/druid/issues/5404

Gian

To unsubscribe from this group and stop receiving emails from it, send an email to druid-user+unsubscribe@googlegroups.com.

To post to this group, send email to druid...@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/druid-user/640aaea5-0bcc-4276-954d-32afd89e6e7c%40googlegroups.com.

[Slim Bouguerra]

@Erwin,

Reading the Document you pointed out I see the following  (https://www.confluent.io/blog/enabling-exactly-kafka-streams/)

"First of all, in Apache Kafka we record offset commits by writing a message to an internal Kafka topic (called the offsets topic). "

Thus am not sure where the gap is coming from? are you sure it is a commit? do you have a reference about how Kafka Streams write commits to the output Stream? 

IMO that will pretty bad Idea anyway. 

Thanks. 

[Erwin Bolwidt]

Hi Slim,

Search for "transaction commit markers" in the first link https://www.confluent.io/blog/transactions-apache-kafka/

I found another resource, the Kafka KIP-98 for transactions:

https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging

Look at section 5.2 of that KIP "5.2 WriteTxnMarkerRequest"