enhancement(nats source): call `drain()` during shutdown (#23635)

* enhancement(nats source): call `drain()` during shutdown

* chore: update changelog

* fix: avoid closing channel prematurely

---------

Co-authored-by: Pavlos Rontidis <pavlos.rontidis@gmail.com>