Hi all, We are debugging an issue with a Kafka Streams application that is producing incorrect output. The application is a simple group-by on a key, and then count. As expected, the application creates a repartitioning topic for the group-by stage. The problem appears to be that messages are getting lost in the repartitioning topic.
Looking at the Kafka broker logs, it appears that the log segments for the repartitioning topic are getting marked for deletion very aggressively (within ~2 seconds of being created), so fast that some segments are deleted before the count stage of the Kafka Streams application has had a chance to consume the messages. I have checked the configuration and I cannot see a reason why the log segments should be getting deleted so quickly. The following line reports the creation of the repartitioning topic: [2018-01-29 15:31:39,992] INFO Created log for partition [streamsapp_site_stats-v290118-debug_1517239899448-localstore_log_event_counts-repartition,0] in /kafka-data with properties {compression.type -> producer, message.format.version -> 0.11.0-IV2, file.delete.delay.ms -> 100000, max.message.bytes -> 1000012, min.compaction.lag.ms -> 0, message.timestamp.type -> CreateTime, min.insync.replicas -> 1, segment.jitter.ms -> 0, preallocate -> false, min.cleanable.dirty.ratio -> 0.5, index.interval.bytes -> 4096, unclean.leader.election.enable -> false, retention.bytes -> 1073741824, delete.retention.ms -> 86400000, cleanup.policy -> delete, flush.ms -> 9223372036854775807, segment.ms -> 3600000, segment.bytes -> 1073741824, retention.ms -> 86400000, message.timestamp.difference.max.ms -> 9223372036854775807, segment.index.bytes -> 10485760, flush.messages -> 9223372036854775807}. (kafka.log.LogManager) As you can see, retention is set to 24 hours, retention size 1 GB, segment rolling time to 1 hour, segment size 1 GB. For test purposes we are running the Streams app on a fixed input of 7,000 messages, with a total size of only about 5.5 MB, so we shouldn't be getting anywhere near the segment or retention limits. The input topic has only one partition. Just two seconds after the topic is created, the broker reports that it is rolling log segments and scheduling old log segments for deletion: [2018-01-29 15:31:41,923] INFO Rolled new log segment for 'streamsapp_site_stats-v290118-debug_1517239899448-localstore_log_event_counts-repartition-0' in 1 ms. (kafka.log.Log) [2018-01-29 15:31:41,924] INFO Scheduling log segment 0 for log streamsapp_site_stats-v290118-debug_1517239899448-localstore_log_event_counts-repartition-0 for deletion. (kafka.log.Log) [2018-01-29 15:31:41,945] INFO Cleared earliest 0 entries from epoch cache based on passed offset 6582 leaving 1 in EpochFile for partition streamsapp_site_stats-v290118-debug_1517239899448-localstore_log_event_counts-repartition-0 (kafka.server.epoch.LeaderEpochFileCache) [2018-01-29 15:31:42,923] INFO Rolled new log segment for 'streamsapp_site_stats-v290118-debug_1517239899448-localstore_log_event_counts-repartition-0' in 2 ms. (kafka.log.Log) [2018-01-29 15:31:42,924] INFO Scheduling log segment 6582 for log streamsapp_site_stats-v290118-debug_1517239899448-localstore_log_event_counts-repartition-0 for deletion. (kafka.log.Log) 100 seconds later (consistent with the setting for file.delete.delay.ms), the files are actually deleted: [2018-01-29 15:33:21,923] INFO Deleting segment 0 from log streamsapp_site_stats-v290118-debug_1517239899448-localstore_log_event_counts-repartition-0. (kafka.log.Log) [2018-01-29 15:33:21,929] INFO Deleting index /kafka-data/streamsapp_site_stats-v290118-debug_1517239899448-localstore_log_event_counts-repartition-0/00000000000000000000.index.deleted (kafka.log.OffsetIndex) [2018-01-29 15:33:21,929] INFO Deleting index /kafka-data/streamsapp_site_stats-v290118-debug_1517239899448-localstore_log_event_counts-repartition-0/00000000000000000000.timeindex.deleted (kafka.log.TimeIndex) [2018-01-29 15:33:22,925] INFO Deleting segment 6582 from log streamsapp_site_stats-v290118-debug_1517239899448-localstore_log_event_counts-repartition-0. (kafka.log.Log) [2018-01-29 15:33:22,926] INFO Deleting index /kafka-data/streamsapp_site_stats-v290118-debug_1517239899448-localstore_log_event_counts-repartition-0/00000000000000006582.index.deleted (kafka.log.OffsetIndex) [2018-01-29 15:33:22,927] INFO Deleting index /kafka-data/streamsapp_site_stats-v290118-debug_1517239899448-localstore_log_event_counts-repartition-0/00000000000000006582.timeindex.deleted (kafka.log.TimeIndex) Does anyone know what might be causing the messages in the repartitioning topic to be deleted so aggressively? Thanks, Martin