add explicit ack/nak for nats jetstream

Signed-off-by: stephen-totty-hpe <stephen.totty@hpe.com>

protocol/nats_jetstream/v2/message.go

@@ -99,6 +99,17 @@ func (m *Message) ReadBinary(ctx context.Context, encoder binding.BinaryWriter)
 
 // Finish *must* be called when message from a Receiver can be forgotten by the receiver.
 func (m *Message) Finish(err error) error {
+	// Ignore error if message has already been Ack/Nak.
+	// Ack and Nak first checks to see if the message has been acknowleged
+	// and immediately returns an error without applying any logic to the message on the server.
+	// A message will auto acknowledge unless manual ack is set or policy is set to AckNonePolicy
+	// see: ManualAck() subscription option
+	// see: ConsumerConfig.AckPolicy
+	if err != nil {
+		_ = m.Msg.Nak()
+		return nil
+	}
+	_ = m.Msg.Ack()
 	return nil
 }