outlet/kafka: execute shutdown before committing work

And add a bit more logging to understand what happens on shutdown.

outlet/core/worker.go

@@ -44,9 +44,11 @@ func (c *Component) newWorker(i int, scaleRequestChan chan<- kafka.ScaleRequest)
 
 // shutdown shutdowns the worker, flushing any remaining data.
 func (w *worker) shutdown() {
+	w.l.Info().Msg("flush final batch to ClickHouse")
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
 	w.cw.Flush(ctx)
+	w.l.Info().Msg("worker stopped")
 }
 
 // processIncomingFlow processes one incoming flow from Kafka.

outlet/kafka/worker.go

@@ -66,6 +66,7 @@ func (c *realComponent) startOneWorker() error {
 			Logger()
 		defer func() {
 			logger.Info().Msg("stopping worker")
+			shutdown()
 
 			// Allow a small grace time to commit uncommited work.
 			ctx, cancelCommit := context.WithTimeout(context.Background(), 10*time.Second)
@@ -74,7 +75,6 @@ func (c *realComponent) startOneWorker() error {
 				logger.Err(err).Msg("cannot commit marked partition offsets")
 			}
 
-			shutdown()
 			client.CloseAllowingRebalance()
 		}()