Make sure to store EOF offset on termination (issue #164)

confluentinc · Nov 17, 2014 · 74be57e · 74be57e
1 parent fb4fff5
commit 74be57e
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/src/rdkafka_offset.c b/src/rdkafka_offset.c
@@ -521,10 +521,10 @@ void rd_kafka_offset_store_term (rd_kafka_toppar_t *rktp) {
         if (!(rktp->rktp_flags & RD_KAFKA_TOPPAR_F_OFFSET_STORE))
                 return;
 
-        if (rktp->rktp_stored_offset == 0 &&
-            rktp->rktp_eof_offset > 0)
+        if (rktp->rktp_stored_offset < rktp->rktp_eof_offset)
                 rd_kafka_offset_store0(rktp, rktp->rktp_eof_offset,
                                        0/*no lock*/);
+
         switch (rktp->rktp_rkt->rkt_conf.offset_store_method)
         {
         case RD_KAFKA_OFFSET_METHOD_FILE: