confluent-kafka-go源码分析：从一个奇怪的offset -1000开始

// #cgo CFLAGS: -DUSE_VENDORED_LIBRDKAFKA -DLIBRDKAFKA_STATICLIB
// #cgo LDFLAGS: ${SRCDIR}/librdkafka_vendor/librdkafka_darwin.a  -lm -lsasl2 -ldl -lpthread
import "C"

// #cgo pkg-config: rdkafka
import "C"

func getConsumer() *kafka.Consumer {
kconsumer, err := kafka.NewConsumer(&kafka.ConfigMap{
    "metadata.broker.list":     groupList,
    "group.id":                 groupID,
    "client.id":                hostname,
    "enable.auto.commit":       false,
    "enable.auto.offset.store": false,
    "auto.offset.reset":        "earliest",
    "enable.partition.eof":     true,
    "debug":                    "msg", //  "generic, broker, topic, metadata, feature, queue, msg, protocol, cgrp, security, fetch, interceptor, plugin, consumer, admin, eos, mock, assignor, conf, all",
  })
  }

func sub(kconsumer *kafka.Consumer) {
  if err := kconsumer.SubscribeTopics([]string{topic}, nil); err != nil {
    fmt.Println(err)
  }
  }

func seek(kconsumer *kafka.Consumer) {
  if err := kconsumer.Assign([]kafka.TopicPartition{{
    Topic:     &topic,
    Partition: 0,
    Offset:    kafka.Offset(offset),
  }}); err != nil {
    fmt.Println(err)
  }


  if err := kconsumer.Seek(kafka.TopicPartition{
    Topic:     &topic,
    Partition: 0,
    Offset:    kafka.Offset(offset),
  }, 400); err != nil {
    fmt.Println(err)
  }
  }

func poll(kconsumer *kafka.Consumer){
for i := 0; i < count; i++ {
      ev := kconsumer.Poll(int(400))
      as, err := kconsumer.Assignment()
      if len(as) > 0 {
        fmt.Println(as, err, int64(as[0].Offset))
      }
      fmt.Println("\033[31m", ev, "\033[m")
      if ev != nil {
        if _, ok := ev.(*kafka.Message); ok {
          fmt.Println("\033[31moffsets&\033[m:", ev.(*kafka.Message).TopicPartition)
        }
      }
    }
}

func main() {
  kconsumer := getConsumer()
  //sub(kconsumer) //情况1
  http.HandleFunc("/", func(http.ResponseWriter, *http.Request) {
    sub(kconsumer)  情况2
    time.Sleep(100 * time.Millisecond) //情况3
    seek(kconsumer)
    poll(kconsumer)
    })
  fmt.Println(http.ListenAndServe(":8088", nil))
}

sh kafka-consumer-groups.sh -bootstrap-server localhost:9092 --group test  --describe


     Consumer group 'test' has no active members.


     GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
     test            test            0          5               9               4               -               -               -

static int rd_kafka_assignment_serve_removals(rd_kafka_t *rk) {
      if (valid_offsets > 0 &&
            rk->rk_conf.offset_store_method == RD_KAFKA_OFFSET_METHOD_BROKER &&
            rk->rk_cgrp && rk->rk_conf.enable_auto_commit &&
            !rd_kafka_destroy_flags_no_consumer_close(rk))
                rd_kafka_cgrp_assigned_offsets_commit(
                    rk->rk_cgrp, rk->rk_consumer.assignment.removed,
                    rd_false /* use offsets from .removed */,
                    "unassigned partitions");

rd_kafka_cgrp_t *rd_kafka_cgrp_new(rd_kafka_t *rk,
                                   const rd_kafkap_str_t *group_id,
                                   const rd_kafkap_str_t *client_id)
  if (rk->rk_conf.enable_auto_commit &&
            rk->rk_conf.auto_commit_interval_ms > 0)
                rd_kafka_timer_start(
                    &rk->rk_timers, &rkcg->rkcg_offset_commit_tmr,
                    rk->rk_conf.auto_commit_interval_ms * 1000ll,
                    rd_kafka_cgrp_offset_commit_tmr_cb, rkcg);

rd_kafka_resp_err_t rd_kafka_offset_store_stop(rd_kafka_toppar_t *rktp) 
    if (rktp->rktp_rkt->rkt_rk->rk_conf.enable_auto_offset_store &&
            rktp->rktp_stored_offset == RD_KAFKA_OFFSET_INVALID &&
            rktp->rktp_offsets_fin.eof_offset > 0)
                rd_kafka_offset_store0(rktp, rktp->rktp_offsets_fin.eof_offset,
                                       rd_true /* force */, RD_DONT_LOCK);

rd_kafka_op_res_t rd_kafka_op_handle_std(rd_kafka_t *rk,
                                         rd_kafka_q_t *rkq,
                                         rd_kafka_op_t *rko,
                                         int cb_type) 
     rd_kafka_fetch_op_app_prepare(rk, rko);
       offset = rko->rko_u.fetch.rkm.rkm_rkmessage.offset + 1;
      if (rk->rk_conf.enable_auto_offset_store)
                rd_kafka_offset_store0(rktp, offset,
                                       /* force: ignore assignment state */
                                       rd_true, RD_DONT_LOCK);z