changefeedccl: fix flush_hist_nanos callback

pkg/ccl/changefeedccl/batching_sink.go

@@ -118,7 +118,6 @@ type rowEvent struct {
 // Flush implements the Sink interface, returning the first error that has
 // occured in the past EmitRow calls.
 func (s *batchingSink) Flush(ctx context.Context) error {
-	defer s.metrics.recordFlushRequestCallback()()
 	flushWaiter := make(chan struct{})
 	select {
 	case <-ctx.Done():

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -793,6 +793,7 @@ func (ca *changeAggregator) flushBufferedEvents() error {
 	if err := ca.eventConsumer.Flush(ca.Ctx()); err != nil {
 		return err
 	}
+	defer ca.sliMetrics.recordFlushRequestCallback()()
 	return ca.sink.Flush(ca.Ctx())
 }
 

pkg/ccl/changefeedccl/sink.go

@@ -582,7 +582,6 @@ func (s *bufferSink) EmitResolvedTimestamp(
 
 // Flush implements the Sink interface.
 func (s *bufferSink) Flush(_ context.Context) error {
-	defer s.metrics.recordFlushRequestCallback()()
 	return nil
 }
 
@@ -679,7 +678,6 @@ func (n *nullSink) EmitResolvedTimestamp(
 
 // Flush implements Sink interface.
 func (n *nullSink) Flush(ctx context.Context) error {
-	defer n.metrics.recordFlushRequestCallback()()
 	if log.V(2) {
 		log.Info(ctx, "flushing")
 	}

pkg/ccl/changefeedccl/sink_cloudstorage.go

@@ -671,8 +671,6 @@ func (s *cloudStorageSink) Flush(ctx context.Context) error {
 		return errors.New(`cannot Flush on a closed sink`)
 	}
 
-	s.metrics.recordFlushRequestCallback()()
-
 	var err error
 	s.files.Ascend(func(i btree.Item) (wantMore bool) {
 		err = s.flushFile(ctx, i.(*cloudStorageSinkFile))
@@ -817,9 +815,7 @@ func (s *cloudStorageSink) asyncFlusher(ctx context.Context) error {
 			}
 
 			// flush file to storage.
-			flushDone := s.metrics.recordFlushRequestCallback()
 			err := req.file.flushToStorage(ctx, s.es, req.dest, s.metrics)
-			flushDone()
 
 			if err != nil {
 				log.Errorf(ctx, "error flushing file to storage: %s", err)

pkg/ccl/changefeedccl/sink_kafka.go

@@ -434,8 +434,6 @@ func (s *kafkaSink) EmitResolvedTimestamp(
 
 // Flush implements the Sink interface.
 func (s *kafkaSink) Flush(ctx context.Context) error {
-	defer s.metrics.recordFlushRequestCallback()()
-
 	flushCh := make(chan struct{}, 1)
 	var inflight int64
 	var flushErr error

pkg/ccl/changefeedccl/sink_pubsub.go

@@ -251,8 +251,6 @@ func (p *deprecatedPubsubSink) Flush(ctx context.Context) error {
 }
 
 func (p *deprecatedPubsubSink) flush(ctx context.Context) error {
-	defer p.metrics.recordFlushRequestCallback()()
-
 	select {
 	case <-ctx.Done():
 		return ctx.Err()

pkg/ccl/changefeedccl/sink_sql.go

@@ -195,8 +195,6 @@ func (s *sqlSink) emit(
 
 // Flush implements the Sink interface.
 func (s *sqlSink) Flush(ctx context.Context) error {
-	defer s.metrics.recordFlushRequestCallback()()
-
 	if len(s.rowBuf) == 0 {
 		return nil
 	}

pkg/ccl/changefeedccl/sink_webhook.go

@@ -690,8 +690,6 @@ func (s *deprecatedWebhookSink) EmitResolvedTimestamp(
 }
 
 func (s *deprecatedWebhookSink) Flush(ctx context.Context) error {
-	s.metrics.recordFlushRequestCallback()()
-
 	// Send flush request.
 	select {
 	case <-ctx.Done():