We should add logic to complete futures during irregular stream shutdown (when subscriber.close() is called, to prevent possible deadlocks.