@@ -512,6 +512,7 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p
512512
513513 start = time .Now ()
514514 var hadPublicationError bool
515+ loop:
515516 for i , e := range events {
516517 event , ok := e .(map [string ]interface {})
517518 if ! ok {
@@ -539,6 +540,25 @@ func (i input) run(env v2.Context, src *source, cursor map[string]interface{}, p
539540 pubCursor = cursor
540541 }
541542 }
543+ // This is checked prior to the publish attempt since the
544+ // cursor.Publisher interface does not document the behaviour
545+ // related to context cancellation and the context is not
546+ // explicitly passed in, so favour this explicit clarity.
547+ switch err := ctx .Err (); {
548+ case err == nil :
549+ case errors .Is (err , context .Canceled ), errors .Is (err , context .DeadlineExceeded ):
550+ log .Infow ("context cancelled with unpublished events" , "unpublished" , len (events )- i )
551+ // Don't update status, since we are about to pass
552+ // through the Running state and then fall through
553+ // to the input exit with a change to Stopped.
554+ break loop
555+ default :
556+ // This should never happen.
557+ log .Warnw ("failed with unpublished events" , "error" , err , "unpublished" , len (events )- i )
558+ health .UpdateStatus (status .Degraded , "error publishing events: " + err .Error ())
559+ isDegraded = true
560+ break loop
561+ }
542562 err = pub .Publish (beat.Event {
543563 Timestamp : time .Now (),
544564 Fields : event ,
0 commit comments