Set context for copied message on retry

[sevaorlov]

Sets the context for the message copy when it is retried.

The problem is that currently if a handler fails to process a message - it is copied, but without a context. That might result in the errors in the handler if it waits for the partition, offset or message timestamp fields in the context, which are set here

watermill-kafka/pkg/kafka/subscriber.go

Lines 546 to 548 in b03a4f4

	ctx = setPartitionToCtx(ctx, kafkaMsg.Partition)
	ctx = setPartitionOffsetToCtx(ctx, kafkaMsg.Offset)
	ctx = setMessageTimestampToCtx(ctx, kafkaMsg.Timestamp)

Context is created here and set for the original message

watermill-kafka/pkg/kafka/subscriber.go

Lines 556 to 558 in b03a4f4

	ctx, cancelCtx := context.WithCancel(ctx)
	msg.SetContext(ctx)
	defer cancelCtx()

[m110]

Thank you @sevaorlov!