Add timeout for producing records

e2e/producer.go

@@ -30,10 +30,10 @@ func (s *Service) produceMessage(ctx context.Context, partition int) {
 	// This childCtx will ensure that we will abort our efforts to produce (including retries) when we exceed
 	// the SLA for producers.
 	childCtx, cancel := context.WithTimeout(ctx, s.config.Producer.AckSla)
-	defer cancel()
 
 	s.endToEndMessagesProducedInFlight.Inc()
 	s.client.Produce(childCtx, record, func(r *kgo.Record, err error) {
+		defer cancel()
 		ackDuration := time.Since(startTime)
 		s.endToEndMessagesProducedInFlight.Dec()
 		s.endToEndMessagesProducedTotal.Inc()