【发布时间】：2016-05-19 15:55:31
【问题描述】：
据我从 Kafka 0.9.0.1 了解，偏移量在 Kafka 的主题中进行管理。 奇怪的是，当消费者组死亡时，偏移量会从主题中删除。下次我从相同的 consumerGroupId 开始时 - 偏移量会从最早的时候重置回来。
这是预期的吗？即使消费者组完全死亡，我也希望偏移量保持不变，并在它重新启动时从它停止的偏移量继续。
这里是consumer.config中的设置：
metric.reporters = []
metadata.max.age.ms = 300000
value.deserializer = classfk.el.client.mappers.kafka.KafkaDeserializer
group.id =testConsumerId
partition.assignment.strategy = [org.apache.kafka.clients.consumer.RangeAssignor]
reconnect.backoff.ms = 50
sasl.kerberos.ticket.renew.window.factor = 0.8
max.partition.fetch.bytes = 1048576
bootstrap.servers = [k1:9092]
retry.backoff.ms = 100
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
ssl.keystore.type = JKS
ssl.trustmanager.algorithm = PKIX
enable.auto.commit = false
ssl.key.password = null
fetch.max.wait.ms = 500
sasl.kerberos.min.time.before.relogin = 60000
connections.max.idle.ms = 540000
ssl.truststore.password = null
session.timeout.ms = 30000
metrics.num.samples = 2
client.id =testAppId1463671497716
ssl.endpoint.identification.algorithm = null
key.deserializer = classorg.apache.kafka.common.serialization.StringDeserializer
ssl.protocol = TLS
check.crcs = true
request.timeout.ms = 40000
ssl.provider = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.keystore.location = null
heartbeat.interval.ms = 3000
auto.commit.interval.ms = 5000
receive.buffer.bytes = 32768
ssl.cipher.suites = null
ssl.truststore.type = JKS
security.protocol = PLAINTEXT
ssl.truststore.location = null
ssl.keystore.password = null
ssl.keymanager.algorithm = SunX509
metrics.sample.window.ms = 30000
fetch.min.bytes = 1
send.buffer.bytes = 131072
auto.offset.reset = earliest
我在日志中看到：
20:55:00.024 调试 o.a.k.c.c.i.AbstractCoordinator - 发行领导 同步组（SYNC_GROUP： {GROUP_ID = testConsumerId，Generation ID的= 1，member_id = testAppId1463671497716-d1ce3669-b451-4197-a5dd-39dd38c61102，group_assignment = [{member_id = testAppId1463671497716-d1ce3669-b451-4197-a5dd-39dd38c61102，member_assignment = java.nio.HeapByteBuffer [POS =0 lim=36 cap=36]}]}）到协调器 2147483647 20:55:00.379 调试 o.a.k.c.c.i.AbstractCoordinator - 收到成功的同步组 组 testConsumerId 的响应： {error_code=0,member_assignment=java.nio.HeapByteBuffer[pos=0 lim=36 cap=36]} 20:55:00.431 调试 o.a.k.c.c.i.ConsumerCoordinator - 设置 新分配的分区 [sampleEntity-1, sampleEntity-0] 20:55:00.432 调试 o.a.k.c.c.i.ConsumerCoordinator - 获取 分区的提交偏移量：[sampleEntity-1，sampleEntity-0] 20:55:00.605 调试 o.a.k.c.c.i.ConsumerCoordinator - 未提交 分区 sampleEntity-1 20:55:00.606 的偏移量 o.a.k.c.consumer.internals.Fetcher - 重置分区偏移量 sampleEntity-1 到最早的偏移量。 20:55:00.732 o.a.k.c.consumer.internals.Fetcher - 获取的分区偏移量 0 sampleEntity-1 20:55:00.732 o.a.k.c.consumer.internals.Fetcher - 将分区 sampleEntity-0 的偏移量重置为提交的偏移量 25
在服务器日志上，当消费者启动和死亡时 - [2016-05-19
16:09:50,113] 信息 [GroupCoordinator 0]：准备重新稳定 将 testConsumerId 与老一代 0 分组 (kafka.coordinator.GroupCoordinator) [2016-05-19 16:09:50,114] 信息 [GroupCoordinator 0]：稳定组 testConsumerId 生成 1 (kafka.coordinator.GroupCoordinator) [2016-05-19 16:09:50,125] 信息 [GroupCoordinator 0]：从组长收到的分配 第 1 代的 testConsumerId (kafka.coordinator.GroupCoordinator) [2016-05-19 16:09:50,158] TRACE [代理 0 上的组元数据管理器]： 获取组的偏移量 Vector([sampleEntity,1], [sampleEntity,0]) 测试消费者 ID。 (kafka.coordinator.GroupMetadataManager) [2016-05-19 16:10:38,158] 跟踪 [GroupCoordinator 0]：成员 组中的testAppId1463674187858-ea8c9c30-4c9d-4b52-bfef-44c299442d45 testConsumerId 失败（kafka.coordinator.GroupCoordinator） [2016-05-19 16:10:38,158] 信息 [GroupCoordinator 0]：准备 使用老一代 1 重新稳定组 testConsumerId (kafka.coordinator.GroupCoordinator) [2016-05-19 16:10:38,158] TRACE [代理 0 上的组元数据管理器]：将组 testConsumerId 标记为 删除。 (kafka.coordinator.GroupMetadataManager) [2016-05-19 16:10:38,159] 信息 [GroupCoordinator 0]：组 testConsumerId 第 1 代已死亡并被移除 (kafka.coordinator.GroupCoordinator)
在它死掉并被移除后，我无法访问之前的偏移量。
【问题讨论】：
-
你是否尝试增加：offsets.retention.minutes
标签： apache-kafka kafka-consumer-api