SubSourceLogic may send Stop twice

[ennru]

When using the ConsumerControl to stop a consumer the Stop message may be sent twice to the KafkaConsumerActor resulting in a dead letter.
Both SubSourceLogic.performShutdown and SubSourceLogic.postStop send Stop.

16:51:08.800 DEBUG a.kafka.internal.SingleSourceLogic - performShutdown sending Stop
16:51:08.847 DEBUG a.kafka.internal.KafkaConsumerActor - received handled message Stop from Actor[akka://Spec/deadLetters]
16:51:08.848 DEBUG a.kafka.internal.KafkaConsumerActor - Stopping Actor[akka://Spec/system/kafka-consumer-5#-912300600]
16:51:08.848 DEBUG a.kafka.internal.SingleSourceLogic - postStop sending Stop
16:51:08.852 DEBUG a.kafka.internal.KafkaConsumerActor - received handled message Subscribe(Set(topic7),ListenerCallbacks(akka.kafka.internal.SingleSourceLogic$$Lambda$1139/181292641@17274548,akka.kafka.internal.SingleSourceLogic$$Lambda$1140/1239809534@20a96f3c)) from Actor[akka://Spec/system/StreamSupervisor-0/$$f#1905692510]
16:51:08.858 INFO  akka.actor.RepointableActorRef - Message [akka.kafka.KafkaConsumerActor$Stop$] without sender to Actor[akka://Spec/system/kafka-consumer-5#-912300600] was not delivered. [5] dead letters encountered. If this is not an expected behavior, then [Actor[akka://Spec/system/kafka-consumer-5#-912300600]] may have terminated unexpectedly, This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
``

[AndreasPresthammer]

This may not be the same issue, but I observe similar behavior with the SingleSourceLogic component.

Using akka-stream-kafka v2.0.5.

BaseSingleSourceLogic.scala:114 will cause the function performShutdown() in SingleSourceLogic.scala:48 which will shut down the consumer actor.

After this is done the postStop function in SingleSourceLogic.scala:44 will send a StopFromStage message to the (already killed) consumer actor, which causes a dead letter message to appear in the logs.

Repro pseudocode:
Consumer.plainSource(..)
.take(1) // we grab one message, and then shutdown the stream - this should trigger the BaseSingleSourceLogic.scala:114
.runWith(Sink.ignore)

Setting the config akka.kafka.consumer.stop-timeout = 0 will force this issue.

Using the default akka.kafka.consumer.stop-timeout = 30 will also reproduce the issue, it's just that in that case it will be the delayed StopFromStage message coming from SingleSourceLogic.scala:67 which will trigger the dead letter message, since the postStop SingleSourceLogic.scala:44 would have run first.

[ennru]

Thank you for sharing your observations. It is the same issue indeed.