[Bug] kafka-http-connector: same message processed by different pod instances

[danyc]

Using fission

client:
  fission/core:
    BuildDate: "2020-10-20T14:43:45Z"
    GitCommit: 52508e19343ef93da4ce8960908c5569f44da225
    Version: 1.11.2
server:
  fission/core:
    BuildDate: "2020-10-20T14:43:45Z"
    GitCommit: 52508e19343ef93da4ce8960908c5569f44da225
    Version: 1.11.2

with keda

  mqt_keda:
    enabled: true
    connector_images:
      kafka:
        image: fission/keda-kafka-http-connector
        tag: 1

I have a MessageQueueTrigger defined:

apiVersion: fission.io/v1
kind: MessageQueueTrigger
metadata:
  creationTimestamp: null
  name: ...-test
  namespace: default
spec:
  contentType: application/json
  cooldownPeriod: 60
  errorTopic: ....errors.v1
  functionref:
    functionweights: null
    name: ...-test
    type: name
  maxReplicaCount: 1
  maxRetries: 3
  messageQueueType: kafka
  metadata:
    bootstrapServers: my-kafka:9092
    consumerGroup: ...-test
    topic: ....requests.v1
    lagThreshold: 100
  minReplicaCount: 0
  mqtkind: keda
  pollingInterval: 30
  respTopic: ....output.v1
  topic: ....requests.v1

Scenario: publish 200 messages on input topic and check keda scaling.

Actual results: when lag > 100, a new consumer pod was started, but some of the messages were processed by both consumers (screenshot attached)

Expected: the consumer should handle kafka re-balancing and ensure single message processing.

Found this issue reported on sarama library: IBM/sarama#1516 and one of the comments mentions:

The above is a symptom of not looping on Consume(). Consume() will exit without error when a rebalancing occurs and it is up to the user to call it again when this occurs.

Under the hood it seems like when a re-balance occurs all sessions are torn down completely (briefly no members exist and therefore no partitions are handled by anyone) and when you re-call Consume() a new session is brought up which should get its share of the partitions.

I'm not familiar with Go. Please check the code and asses if there is room for improvement regarding the consumer handling.
Additionally, maybe update sarama library to latest version.