Backport #1452 (#1466)

* Fix issue with broker dropping messages with "No TTL Seen"

* Treat the broker TTL attribute as case insensitive, in line with how
cloudevents attributes should be handled. See
  cloudevents/spec#321
for context.

* Remove unnecessary test change.

pkg/broker/receiver.go

@@ -126,8 +126,8 @@ func (r *Receiver) serveHTTP(ctx context.Context, event cloudevents.Event, resp
 
 	// Remove the TTL attribute that is used by the Broker.
 	originalV2 := event.Context.AsV02()
-	ttl, present := originalV2.Extensions[V02TTLAttribute]
-	if !present {
+	ttl, ttlKey := GetTTL(event.Context)
+	if ttl == nil {
 		// Only messages sent by the Broker should be here. If the attribute isn't here, then the
 		// event wasn't sent by the Broker, so we can drop it.
 		r.logger.Warn("No TTL seen, dropping", zap.Any("triggerRef", triggerRef), zap.Any("event", event))
@@ -136,7 +136,7 @@ func (r *Receiver) serveHTTP(ctx context.Context, event cloudevents.Event, resp
 		// framework returns a 500 to the caller, so the Channel would send this repeatedly.
 		return nil
 	}
-	delete(originalV2.Extensions, V02TTLAttribute)
+	delete(originalV2.Extensions, ttlKey)
 	event.Context = originalV2
 
 	r.logger.Debug("Received message", zap.Any("triggerRef", triggerRef))

pkg/broker/ttl.go

@@ -17,6 +17,8 @@
 package broker
 
 import (
+	"strings"
+
 	cloudevents "github.com/cloudevents/sdk-go"
 )
 
@@ -26,6 +28,18 @@ const (
 	V02TTLAttribute = "knativebrokerttl"
 )
 
+// GetTTL finds the TTL in the EventContext using a case insensitive comparison
+// for the key. The second return param, is the case preserved key that matched.
+// Depending on the encoding/transport, the extension case could be changed.
+func GetTTL(ctx cloudevents.EventContext) (interface{}, string) {
+	for k, v := range ctx.AsV02().Extensions {
+		if lower := strings.ToLower(k); lower == V02TTLAttribute {
+			return v, k
+		}
+	}
+	return nil, V02TTLAttribute
+}
+
 // SetTTL sets the TTL into the EventContext. ttl should be a positive integer.
 func SetTTL(ctx cloudevents.EventContext, ttl interface{}) (cloudevents.EventContext, error) {
 	err := ctx.SetExtension(V02TTLAttribute, ttl)