mqtt-streaming: Partially revert #1818

[longshorej]

Hi! With the changes made in #1818, we're seeing an issue with the MQTT connector. Reverting the 1 line change fixes it. I don't know what the semantics of Behavior.interpretSignal are but the change is causing the following:

2019-07-17T22:13:18.792Z [22:13:18.792UTC] 127.0.0.1 com.cisco.streambed.durablequeue.remote ERROR ActorAdapter - deferred [Deferred(TimerSchedulerImpl.scala:28)] should not be passed to interpreter
java.lang.IllegalArgumentException: deferred [Deferred(TimerSchedulerImpl.scala:28)] should not be passed to interpreter
	at akka.actor.typed.Behavior$.interpret(Behavior.scala:422)
	at akka.actor.typed.Behavior$.interpretSignal(Behavior.scala:399)
	at akka.stream.alpakka.mqtt.streaming.impl.QueueOfferState$$anonfun$waitForQueueOfferCompleted$3.applyOrElse(QueueOfferState.scala:46)
	at akka.stream.alpakka.mqtt.streaming.impl.QueueOfferState$$anonfun$waitForQueueOfferCompleted$3.applyOrElse(QueueOfferState.scala:46)
	at akka.actor.typed.internal.BehaviorImpl$ReceiveBehavior.receiveSignal(BehaviorImpl.scala:35)
	at akka.actor.typed.Behavior$.interpret(Behavior.scala:436)
	at akka.actor.typed.Behavior$.interpretSignal(Behavior.scala:399)
	at akka.actor.typed.internal.adapter.ActorAdapter.handleSignal(ActorAdapter.scala:128)
	at akka.actor.typed.internal.adapter.ActorAdapter.aroundReceive(ActorAdapter.scala:87)
	at akka.actor.ActorCell.receiveMessage(ActorCell.scala:612)
	at akka.actor.dungeon.DeathWatch.$anonfun$receivedTerminated$1(DeathWatch.scala:67)
	at akka.actor.dungeon.DeathWatch.$anonfun$receivedTerminated$1$adapted(DeathWatch.scala:65)
	at scala.Option.foreach(Option.scala:274)
	at akka.actor.dungeon.DeathWatch.receivedTerminated(DeathWatch.scala:65)
	at akka.actor.dungeon.DeathWatch.receivedTerminated$(DeathWatch.scala:64)
	at akka.actor.ActorCell.receivedTerminated(ActorCell.scala:447)
	at akka.actor.ActorCell.autoReceiveMessage(ActorCell.scala:597)
	at akka.actor.ActorCell.invoke(ActorCell.scala:580)
	at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:268)
	at akka.dispatch.Mailbox.run(Mailbox.scala:229)
	at akka.dispatch.Mailbox.exec(Mailbox.scala:241)
	at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
	at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
	at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
	at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

We're currently rolling out this PR to production as initial tests are looking good, but I wanted to open up a dialog about it. I'll be out for the next couple weeks, but @huntc will further any conversation here.

[2m]

Interesting. We need to add a test-case that would catch the error with the interpretSignal. And then come-up with a different fix for the deprecated Behaviors.orElse. //cc @chbatey @raboof ?

[huntc]

Hey guys - what would you like to do about this?

[huntc]

Any chance we can get this PR merged? The bug appears quite quickly and it will ruin other's experience of the library.

[2m]

Sorry for the delay. I'll merge this, but we will need to figure out a different workaround, since Behavior.orElse is deprecated in 2.6 and will fail the nightly Alpakka build.

[huntc]

Thanks Martynas.

I’ve no idea what the work around should be. Is this one for the Akka core team?