             return
         }
         
-        if !flushNow() {
+        // This flush is triggered by the user and so we should mark all pending messages as flushed
+        if !flushNow(markFlushCheckpoint: true) {
             guard open else {
                 return
             }

                    // Directly update nodes as a promise may trigger a callback that will access the PendingWrites class.
                    updateNodes(pending: pending)
                    
                    // buffer was completely written
                    pending.promise.succeed(result: ())
                    if (isEmpty) {
                        // return nil to signal to the caller that there are no more buffers to consume
                        return nil
                    }

                outstanding -= UInt64(written)
                assert(head != nil)
                
                var allFlushed = false
                var w = written
                while let p = head {
                    assert(!allFlushed)
                    
                    if w >= p.buffer.readableBytes {
                        w -= p.buffer.readableBytes
                        
                        // Directly update nodes as a promise may trigger a callback that will access the PendingWrites class.
                        updateNodes(pending: p)
                        
                        allFlushed = checkFlushCheckpoint(pending: p)

                        // buffer was completely written
                        p.promise.succeed(result: ())
                    } else {
