Discard messages for toppars that saw a seek operation during a fetch or batch processing

src/consumer/__tests__/consumeMessages.spec.js

@@ -461,6 +461,57 @@ describe('Consumer', () => {
     expect(calls).toEqual(1)
   })
 
+  it('skips messages fetched while seek was called', async () => {
+    consumer = createConsumer({
+      cluster: createCluster(),
+      groupId,
+      maxWaitTimeInMs: 1000,
+      logger: newLogger(),
+    })
+
+    const messages = Array(10)
+      .fill()
+      .map(() => {
+        const value = secureRandom()
+        return { key: `key-${value}`, value: `value-${value}` }
+      })
+    await producer.connect()
+    await producer.send({ acks: 1, topic: topicName, messages })
+
+    await consumer.connect()
+
+    await consumer.subscribe({ topic: topicName, fromBeginning: true })
+
+    const sleep = value => waitFor(delay => delay >= value)
+    let offsetsConsumed = []
+
+    const eachBatch = async ({ batch, heartbeat }) => {
+      for (const message of batch.messages) {
+        offsetsConsumed.push(message.offset)
+      }
+
+      await heartbeat()
+    }
+
+    consumer.run({
+      eachBatch,
+    })
+
+    await waitForConsumerToJoinGroup(consumer)
+
+    await waitFor(() => offsetsConsumed.length === messages.length, { delay: 50 })
+    await sleep(50)
+
+    // Hope that we're now in an active fetch state? Something like FETCH_START might help
+    const seekedOffset = offsetsConsumed[Math.floor(messages.length / 2)]
+    consumer.seek({ topic: topicName, partition: 0, offset: seekedOffset })
+    await producer.send({ acks: 1, topic: topicName, messages }) // trigger completion of fetch
+
+    await waitFor(() => offsetsConsumed.length > messages.length, { delay: 50 })
+
+    expect(offsetsConsumed[messages.length]).toEqual(seekedOffset)
+  })
+
   describe('transactions', () => {
     testIfKafka_0_11('accepts messages from an idempotent producer', async () => {
       cluster = createCluster({ allowExperimentalV011: true })

src/consumer/consumerGroup.js

@@ -374,15 +374,17 @@ module.exports = class ConsumerGroup {
             ({ topic }) => topic === topicName
           )
 
-          return partitions.map(partitionData => {
-            const partitionRequestData = topicRequestData.partitions.find(
-              ({ partition }) => partition === partitionData.partition
-            )
+          return partitions
+            .filter(partitionData => !this.seekOffset.has(topicName, partitionData.partition))
+            .map(partitionData => {
+              const partitionRequestData = topicRequestData.partitions.find(
+                ({ partition }) => partition === partitionData.partition
+              )
 
-            const fetchedOffset = partitionRequestData.fetchOffset
+              const fetchedOffset = partitionRequestData.fetchOffset
 
-            return new Batch(topicName, fetchedOffset, partitionData)
-          })
+              return new Batch(topicName, fetchedOffset, partitionData)
+            })
         })
 
         return flatten(batchesPerPartition)
@@ -484,4 +486,8 @@ module.exports = class ConsumerGroup {
       }
     }
   }
+
+  hasSeekOffset({ topic, partition }) {
+    return this.seekOffset.has(topic, partition)
+  }
 }

src/consumer/runner.js

@@ -135,7 +135,7 @@ module.exports = class Runner {
     const { topic, partition } = batch
 
     for (let message of batch.messages) {
-      if (!this.running) {
+      if (!this.running || this.consumerGroup.hasSeekOffset({ topic, partition })) {
         break
       }
 
@@ -187,6 +187,7 @@ module.exports = class Runner {
         },
         uncommittedOffsets: () => this.consumerGroup.uncommittedOffsets(),
         isRunning: () => this.running,
+        isStale: ({ topic, partition }) => this.consumerGroup.hasSeekOffset({ topic, partition }),
       })
     } catch (e) {
       if (!isKafkaJSError(e)) {

src/consumer/seekOffsets.js

@@ -3,6 +3,10 @@ module.exports = class SeekOffsets extends Map {
     super.set([topic, partition], offset)
   }
 
+  has(topic, partition) {
+    return Array.from(this.keys()).some(([t, p]) => t === topic && p === partition)
+  }
+
   pop() {
     if (this.size === 0) {
       return