Add support for cascading retries

events/config.go

@@ -2,12 +2,14 @@ package events
 
 import (
 	"errors"
+
 	"github.com/Shopify/sarama"
 	"github.com/kelseyhightower/envconfig"
 )
 
 const DeadLetterSuffix = "-dead-letters"
 
+// CloudEventsConfig describes the configuration for a consumer of cloud events from a Kafka topic
 type CloudEventsConfig struct {
 	EventSource           string   `envconfig:"CLOUD_EVENTS_SOURCE" required:"true"`
 	KafkaBrokers          []string `envconfig:"KAFKA_BROKERS" required:"true"`
@@ -19,6 +21,9 @@ type CloudEventsConfig struct {
 	KafkaVersion          string   `envconfig:"KAFKA_VERSION" required:"true"`
 	KafkaUsername         string   `envconfig:"KAFKA_USERNAME" required:"false"`
 	KafkaPassword         string   `envconfig:"KAFKA_PASSWORD" required:"false"`
+	KafkaDelay            int      `envconfig:"KAFKA_DELAY" default:"0"`
+	CascadeDelays         []int    `envconfig:"KAFKA_CASCADE_DELAYS" default:""`
+	CascadePattern        string   `envconfig:"KAFKA_CASCADE_PATTERN" default:"%s-delay-%d"`
 	SaramaConfig          *sarama.Config
 }
 

events/consumer.go

@@ -2,12 +2,15 @@ package events
 
 import (
 	"context"
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
 	"github.com/Shopify/sarama"
 	"github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2"
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/cloudevents/sdk-go/v2/binding"
-	"log"
-	"sync"
 )
 
 type SaramaConsumer struct {
@@ -19,6 +22,62 @@ type SaramaConsumer struct {
 	deadLetterProducer *KafkaCloudEventsProducer
 }
 
+//CascadingEventConsumer an event consumer that cascaded failures
+type CascadingEventConsumer struct {
+	Consumers []EventConsumer
+}
+
+// NewCascadingCloudEventsConsumer create a cascading events consumer
+func NewCascadingCloudEventsConsumer(config *CloudEventsConfig) (EventConsumer, error) {
+	topic := config.KafkaTopic
+	delay := config.KafkaDelay
+	var consumers []EventConsumer
+	for nextDelay := range config.CascadeDelays {
+		newconfig := *config
+		newconfig.KafkaDelay = delay
+		newconfig.KafkaTopic = topic
+		newconfig.KafkaDeadLettersTopic = fmt.Sprintf(config.CascadePattern, config.KafkaTopic, nextDelay)
+		consumer, err := NewSaramaCloudEventsConsumer(&newconfig)
+		if err != nil {
+			return nil, err
+		}
+		consumers = append(consumers, consumer)
+		topic = newconfig.KafkaDeadLettersTopic
+		delay = nextDelay
+	}
+	newconfig := *config
+	newconfig.KafkaDelay = delay
+	newconfig.KafkaTopic = topic
+	consumer, err := NewSaramaCloudEventsConsumer(&newconfig)
+	if err != nil {
+		return nil, err
+	}
+	consumers = append(consumers, consumer)
+
+	return &CascadingEventConsumer{
+		Consumers: consumers,
+	}, nil
+}
+
+//RegisterHandler registers the handler with all consumers in the cascade
+func (c *CascadingEventConsumer) RegisterHandler(handler EventHandler) {
+	for _, consumer := range c.Consumers {
+		consumer.RegisterHandler(handler)
+	}
+}
+
+//Start starts an async consumer goproc (that may sleep)
+func (c *CascadingEventConsumer) Start(ctx context.Context) error {
+	for _, consumer := range c.Consumers {
+		err := consumer.Start(ctx)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+//NewSaramaCloudEventsConsumer creates a new cloud events consumer
 func NewSaramaCloudEventsConsumer(config *CloudEventsConfig) (EventConsumer, error) {
 	if err := validateConsumerConfig(config); err != nil {
 		return nil, err
@@ -32,19 +91,36 @@ func NewSaramaCloudEventsConsumer(config *CloudEventsConfig) (EventConsumer, err
 	}, nil
 }
 
+//Setup marks a consumer to be ready
 func (s *SaramaConsumer) Setup(session sarama.ConsumerGroupSession) error {
 	// Mark the consumer as ready
 	close(s.ready)
 	return nil
 }
 
+//Cleanup frees any resources
 func (s *SaramaConsumer) Cleanup(session sarama.ConsumerGroupSession) error {
 	return nil
 }
 
+//ConsumeClaim synchronously consumes all the messages in a claim, optionally delaying consumption a fixed amount of time
 func (s *SaramaConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+	delay := time.Second * time.Duration(s.config.KafkaDelay)
 	for message := range claim.Messages() {
 		m := kafka_sarama.NewMessageFromConsumerMessage(message)
+		timestamp := message.Timestamp
+		scheduledTime := timestamp.Add(delay)
+		remaining := time.Now().Sub(scheduledTime)
+		if delay == 0 || remaining > 0 {
+			timer := time.NewTimer(remaining)
+			select {
+			case <-timer.C:
+				break
+			case <-session.Context().Done():
+				timer.Stop()
+				return nil
+			}
+		}
 		// just ignore non-cloud event messages
 		if rs, rserr := binding.ToEvent(context.Background(), m); rserr == nil {
 			s.handleCloudEvent(*rs)
@@ -76,10 +152,12 @@ func (s *SaramaConsumer) sendToDeadLetterTopic(ce cloudevents.Event) {
 	}
 }
 
+//RegisterHandler register a handler to process events
 func (s *SaramaConsumer) RegisterHandler(handler EventHandler) {
 	s.handlers = append(s.handlers, handler)
 }
 
+//Start starts an async consumer goproc (that may sleep)
 func (s *SaramaConsumer) Start(ctx context.Context) error {
 	if err := s.initialize(); err != nil {
 		return err