Fix race condition in message tracker

Author: weeco

e2e/endtoend_message.go

@@ -14,8 +14,9 @@ type EndToEndMessage struct {
 	Timestamp int64  `json:"createdUtcNs"` // when the message was created, unix nanoseconds
 
 	// The following properties are only used within the message tracker
-	partition int
-	state     int
+	partition      int
+	state          int
+	produceLatency float64
 }
 
 func (m *EndToEndMessage) creationTime() time.Time {

e2e/message_tracker.go

@@ -66,7 +66,11 @@ func (t *messageTracker) updateItemIfExists(msg *EndToEndMessage) error {
 		return ttlcache.ErrNotFound
 	}
 
-	t.cache.SetWithTTL(msg.MessageID, msg, remainingTTL)
+	err = t.cache.SetWithTTL(msg.MessageID, msg, remainingTTL)
+	if err != nil {
+		panic(err)
+	}
+
 	return nil
 }
 
@@ -129,5 +133,6 @@ func (t *messageTracker) onMessageExpired(_ string, reason ttlcache.EvictionReas
 		zap.Int("partition", msg.partition),
 		zap.String("message_id", msg.MessageID),
 		zap.Bool("successfully_produced", msg.state == EndToEndMessageStateProducedSuccessfully),
+		zap.Float64("produce_latency_seconds", msg.produceLatency),
 	)
 }

e2e/producer.go

@@ -58,7 +58,12 @@ func (s *Service) produceMessage(ctx context.Context, partition int) {
 			// We need to use updateItemIfExists() because it's possible that the message has already been consumed
 			// before we have received the message here (because we were awaiting the produce ack).
 			msg.state = EndToEndMessageStateProducedSuccessfully
-			s.messageTracker.updateItemIfExists(msg)
+			msg.produceLatency = ackDuration.Seconds()
+
+			// TODO: Enable again as soon as https://github.com/ReneKroon/ttlcache/issues/60 is fixed
+			// Because we cannot update cache items in an atomic fashion we currently can't use this method
+			// as this would cause a race condition which ends up in records being reported as lost/expired.
+			// s.messageTracker.updateItemIfExists(msg)
 		}
 
 		s.produceLatency.WithLabelValues(pID).Observe(ackDuration.Seconds())