Repository: triggermesh/aws-event-sources (go)

Issue:
Agressive polling causes rate limit hits
All AWS sources poll AWS APIs in a `for` loop, which causes a lot of issues due to rate limiting.

At the very least we should throttle our polling with a [`time.Ticker`](https://golang.org/pkg/time/#Ticker).

Pull request description:
Exponential backoff for receiver poll loops
- got rid of events custom type in favor of complete AWS event body
- adapter's `Start()` method uses `backoff.Run()` function to invoke receiver logic with exponential backoff duration, closes #139


Patch:
diff --git a/pkg/adapter/awscodecommitsource/adapter.go b/pkg/adapter/awscodecommitsource/adapter.go
index f1940d5..9ec5fb5 100644
--- a/pkg/adapter/awscodecommitsource/adapter.go
+++ b/pkg/adapter/awscodecommitsource/adapter.go
@@ -140,12 +140,16 @@ func (a *adapter) Start(stopCh <-chan struct{}) error {
 		a.logger.Errorw("Failed to process pull requests", "error", err)
 	}
 
-	//range time.Tick(time.Duration(syncTime) * time.Second)
-	for {
+	backoff := common.NewBackoff()
+
+	err = backoff.Run(stopCh, func(ctx context.Context) (bool, error) {
+		resetBackoff := false
+
 		if strings.Contains(a.gitEvents, pushEventType) {
 			err := a.processCommits()
 			if err != nil {
 				a.logger.Errorw("Failed to process commits", "error", err)
+				return resetBackoff, nil
 			}
 		}
 
@@ -153,19 +157,25 @@ func (a *adapter) Start(stopCh <-chan struct{}) error {
 			pullRequests, err := a.preparePullRequests()
 			if err != nil {
 				a.logger.Errorw("Failed to process pull requests", "error", err)
+				return resetBackoff, nil
 			}
 
 			pullRequests = removeOldPRs(processedPullRequests, pullRequests)
 
 			for _, pr := range pullRequests {
+				resetBackoff = true
 				err = a.sendPREvent(pr)
 				if err != nil {
 					a.logger.Errorw("Failed to send PR event", "error", err)
+					return resetBackoff, nil
 				}
 				processedPullRequests = append(processedPullRequests, pr)
 			}
 		}
-	}
+		return resetBackoff, nil
+	})
+
+	return err
 }
 
 func (a *adapter) processCommits() error {
@@ -243,20 +253,12 @@ func (a *adapter) preparePullRequests() ([]*codecommit.PullRequest, error) {
 func (a *adapter) sendPushEvent(commit *codecommit.Commit) error {
 	a.logger.Info("Sending Push event")
 
-	data := &PushEvent{
-		Commit:           commit,
-		CommitRepository: &a.arn.Resource,
-		CommitBranch:     &a.branch,
-		EventSource:      aws.String("aws:codecommit"),
-		AwsRegion:        &a.arn.Region,
-	}
-
 	event := cloudevents.NewEvent(cloudevents.VersionV1)
 	event.SetType(v1alpha1.AWSEventType(a.arn.Service, pushEventType))
 	event.SetSubject(fmt.Sprintf("%s/%s", a.arn.Resource, a.branch))
 	event.SetSource(a.arn.String())
 	event.SetID(*commit.CommitId)
-	if err := event.SetData(cloudevents.ApplicationJSON, data); err != nil {
+	if err := event.SetData(cloudevents.ApplicationJSON, commit); err != nil {
 		return fmt.Errorf("failed to set event data: %w", err)
 	}
 
@@ -269,20 +271,12 @@ func (a *adapter) sendPushEvent(commit *codecommit.Commit) error {
 func (a *adapter) sendPREvent(pullRequest *codecommit.PullRequest) error {
 	a.logger.Info("Sending Pull Request event")
 
-	data := &PullRequestEvent{
-		PullRequest: pullRequest,
-		Repository:  &a.arn.Resource,
-		Branch:      &a.branch,
-		EventSource: aws.String("aws:codecommit"),
-		AwsRegion:   &a.arn.Region,
-	}
-
 	event := cloudevents.NewEvent(cloudevents.VersionV1)
 	event.SetType(v1alpha1.AWSEventType(a.arn.Service, prEventType))
 	event.SetSubject(a.branch)
 	event.SetSource(a.arn.String())
 	event.SetID(*pullRequest.PullRequestId)
-	if err := event.SetData(cloudevents.ApplicationJSON, data); err != nil {
+	if err := event.SetData(cloudevents.ApplicationJSON, pullRequest); err != nil {
 		return fmt.Errorf("failed to set event data: %w", err)
 	}
 
diff --git a/pkg/adapter/awscodecommitsource/events.go b/pkg/adapter/awscodecommitsource/events.go
deleted file mode 100644
index 8afdbf6..0000000
--- a/pkg/adapter/awscodecommitsource/events.go
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
-Copyright (c) 2019-2020 TriggerMesh Inc.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-   http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package awscodecommitsource
-
-import "github.com/aws/aws-sdk-go/service/codecommit"
-
-// PushEvent represent a push event from CodeCommit source.
-type PushEvent struct {
-	Commit           *codecommit.Commit `json:"commit"`
-	CommitRepository *string            `json:"commitRepository"`
-	CommitBranch     *string            `json:"commitBranch"`
-	CommitHash       *string            `json:"commitHash"`
-	EventSource      *string            `json:"eventSource"`
-	AwsRegion        *string            `json:"awsRegion"`
-}
-
-// PullRequestEvent represent a PR event from CodeCommit source.
-type PullRequestEvent struct {
-	PullRequest *codecommit.PullRequest `json:"pullRequest"`
-	EventType   *string                 `json:"eventType"`
-	Repository  *string                 `json:"repository"`
-	Branch      *string                 `json:"branch"`
-	EventSource *string                 `json:"eventSource"`
-	AwsRegion   *string                 `json:"awsRegion"`
-}
diff --git a/pkg/adapter/awscognitosource/adapter.go b/pkg/adapter/awscognitosource/adapter.go
index b1de130..f47d5f7 100644
--- a/pkg/adapter/awscognitosource/adapter.go
+++ b/pkg/adapter/awscognitosource/adapter.go
@@ -93,7 +93,10 @@ func NewAdapter(ctx context.Context, envAcc pkgadapter.EnvConfigAccessor, ceClie
 func (a *adapter) Start(stopCh <-chan struct{}) error {
 	a.logger.Infof("Listening to AWS Cognito stream for Identity: %s", a.identityPoolID)
 
-	for {
+	backoff := common.NewBackoff()
+
+	err := backoff.Run(stopCh, func(ctx context.Context) (bool, error) {
+		resetBackoff := false
 		identities, err := a.getIdentities()
 		if err != nil {
 			a.logger.Error(err)
@@ -105,6 +108,7 @@ func (a *adapter) Start(stopCh <-chan struct{}) error {
 		}
 
 		for _, dataset := range datasets {
+			resetBackoff = true
 			records, err := a.getRecords(dataset)
 			if err != nil {
 				a.logger.Error(err)
@@ -116,7 +120,10 @@ func (a *adapter) Start(stopCh <-chan struct{}) error {
 				a.logger.Errorf("SendCloudEvent failed: %v", err)
 			}
 		}
-	}
+		return resetBackoff, nil
+	})
+
+	return err
 }
 
 func (a *adapter) getIdentities() ([]*cognitoidentity.IdentityDescription, error) {
diff --git a/pkg/adapter/awscognitouserpoolsource/adapter.go b/pkg/adapter/awscognitouserpoolsource/adapter.go
index 9ff8d91..b0cae46 100644
--- a/pkg/adapter/awscognitouserpoolsource/adapter.go
+++ b/pkg/adapter/awscognitouserpoolsource/adapter.go
@@ -39,12 +39,6 @@ import (
 	"github.com/triggermesh/aws-event-sources/pkg/apis/sources/v1alpha1"
 )
 
-var (
-	// number of seconds to wait between poll requests
-	// TODO: replace with exponential backoff
-	pollInterval = 5
-)
-
 // envConfig is a set parameters sourced from the environment for the source's
 // adapter.
 type envConfig struct {
@@ -99,31 +93,30 @@ func (a *adapter) Start(stopCh <-chan struct{}) error {
 
 	var latestTimestamp time.Time
 
-	ticker := time.NewTicker(time.Duration(pollInterval) * time.Second)
-	defer ticker.Stop()
+	backoff := common.NewBackoff()
 
-	for {
-		select {
-		case <-stopCh:
-			a.logger.Info("Exiting receiver loop")
-			return nil
-		case <-ticker.C:
-			users, err := a.listUsers()
-			if err != nil {
-				a.logger.Errorf("Cognito ListUsers failed: %v", err)
-				continue
-			}
+	err := backoff.Run(stopCh, func(ctx context.Context) (bool, error) {
+		resetBackoff := false
+		users, err := a.listUsers()
+		if err != nil {
+			a.logger.Errorf("Cognito ListUsers failed: %v", err)
+			return resetBackoff, err
+		}
 
-			users, latestTimestamp = filterByTimestamp(users, latestTimestamp)
+		users, latestTimestamp = filterByTimestamp(users, latestTimestamp)
 
-			for _, user := range users {
-				err := a.sendCognitoEvent(user)
-				if err != nil {
-					a.logger.Errorf("Failed to send cloudevent: %v", err)
-				}
+		for _, user := range users {
+			// we have new users - reset backoff duration
+			resetBackoff = true
+			err := a.sendCognitoEvent(user)
+			if err != nil {
+				a.logger.Errorf("Failed to send cloudevent: %v", err)
 			}
 		}
-	}
+		return resetBackoff, nil
+	})
+
+	return err
 }
 
 func (a *adapter) listUsers() ([]*cognitoidentityprovider.UserType, error) {
diff --git a/pkg/adapter/awsdynamodbsource/adapter.go b/pkg/adapter/awsdynamodbsource/adapter.go
index b0b96c4..0ed1f7b 100644
--- a/pkg/adapter/awsdynamodbsource/adapter.go
+++ b/pkg/adapter/awsdynamodbsource/adapter.go
@@ -112,19 +112,35 @@ func (a *adapter) Start(stopCh <-chan struct{}) error {
 
 	a.logger.Debugf("Streams descriptions: %v", streamsDescriptions)
 
-	for {
+	backoff := common.NewBackoff()
+
+	err = backoff.Run(stopCh, func(ctx context.Context) (bool, error) {
+		resetBackoff := false
 		shardIterators, err := a.getShardIterators(streamsDescriptions)
 		if err != nil {
 			a.logger.Errorw("Failed to get shard iterators", zap.Error(err))
 		}
 
+		var records []*dynamodbstreams.Record
 		for _, shardIterator := range shardIterators {
-			if err := a.processLatestRecords(shardIterator); err != nil {
+			shardRecords, err := a.processLatestRecords(shardIterator)
+			if err != nil {
 				a.logger.Errorw("Error while processing records for shard iterator "+
 					*shardIterator, zap.Error(err))
 			}
+			records = append(records, shardRecords...)
 		}
-	}
+
+		for _, record := range records {
+			resetBackoff = true
+			if err := a.sendDynamoDBEvent(record); err != nil {
+				a.logger.Errorw("Failed to send CloudEvent", zap.Error(err))
+			}
+		}
+		return resetBackoff, nil
+	})
+
+	return err
 }
 
 func (a *adapter) getStreams() ([]*dynamodbstreams.Stream, error) {
@@ -193,23 +209,17 @@ func (a *adapter) getShardIterators(streamsDescriptions []*dynamodbstreams.Strea
 	return shardIterators, nil
 }
 
-func (a *adapter) processLatestRecords(shardIterator *string) error {
+func (a *adapter) processLatestRecords(shardIterator *string) ([]*dynamodbstreams.Record, error) {
 	getRecordsInput := dynamodbstreams.GetRecordsInput{
 		ShardIterator: shardIterator,
 	}
 
 	getRecordsOutput, err := a.dyndbClient.GetRecords(&getRecordsInput)
 	if err != nil {
-		return fmt.Errorf("failed to get records: %w", err)
-	}
-
-	for _, record := range getRecordsOutput.Records {
-		if err := a.sendDynamoDBEvent(record); err != nil {
-			a.logger.Errorw("Failed to send CloudEvent", zap.Error(err))
-		}
+		return []*dynamodbstreams.Record{}, fmt.Errorf("failed to get records: %w", err)
 	}
 
-	return nil
+	return getRecordsOutput.Records, nil
 }
 
 func (a *adapter) sendDynamoDBEvent(record *dynamodbstreams.Record) error {
diff --git a/pkg/adapter/awskinesissource/adapter.go b/pkg/adapter/awskinesissource/adapter.go
index 6c7f1c2..47641de 100644
--- a/pkg/adapter/awskinesissource/adapter.go
+++ b/pkg/adapter/awskinesissource/adapter.go
@@ -102,19 +102,32 @@ func (a *adapter) Start(stopCh <-chan struct{}) error {
 	a.logger.Infof("Connected to Kinesis stream: %s", *streamARN)
 
 	// Obtain records inputs for different shards
-	inputs, shardIDs := a.getRecordsInputs(myStream.StreamDescription.Shards)
+	inputs := a.getRecordsInputs(myStream.StreamDescription.Shards)
 
-	for {
-		err := a.processInputs(inputs, shardIDs, streamARN)
+	backoff := common.NewBackoff()
+
+	err = backoff.Run(stopCh, func(ctx context.Context) (bool, error) {
+		resetBackoff := false
+		records, err := a.processInputs(inputs)
 		if err != nil {
-			a.logger.Errorw("Failed to process inputs", zap.Error(err))
+			a.logger.Errorw("There were errors during inputs processing", zap.Error(err))
 		}
-	}
+
+		for _, record := range records {
+			resetBackoff = true
+			err = a.sendKinesisRecord(record)
+			if err != nil {
+				a.logger.Errorw("Failed to send cloudevent", zap.Error(err))
+			}
+		}
+		return resetBackoff, nil
+	})
+
+	return err
 }
 
-func (a *adapter) getRecordsInputs(shards []*kinesis.Shard) ([]kinesis.GetRecordsInput, []*string) {
+func (a *adapter) getRecordsInputs(shards []*kinesis.Shard) []kinesis.GetRecordsInput {
 	inputs := []kinesis.GetRecordsInput{}
-	shardIDs := []*string{}
 
 	// Kinesis stream might have several shards and each of them had "LATEST" Iterator.
 	for _, shard := range shards {
@@ -136,33 +149,25 @@ func (a *adapter) getRecordsInputs(shards []*kinesis.Shard) ([]kinesis.GetRecord
 		}
 
 		inputs = append(inputs, input)
-		shardIDs = append(shardIDs, shard.ShardId)
 	}
 
-	return inputs, shardIDs
+	return inputs
 }
 
-func (a *adapter) processInputs(inputs []kinesis.GetRecordsInput, shardIDs []*string, streamARN *string) error {
+func (a *adapter) processInputs(inputs []kinesis.GetRecordsInput) ([]*kinesis.Record, error) {
 	var errs []error
+	records := []*kinesis.Record{}
 
 	for i, input := range inputs {
 		input := input
 
 		recordsOutput, err := a.knsClient.GetRecords(&input)
 		if err != nil {
-			a.logger.Errorw("Failed to get records", zap.Error(err))
 			errs = append(errs, err)
 			continue
 		}
 
-		shardID := shardIDs[i]
-
-		for _, record := range recordsOutput.Records {
-			if err := a.sendKinesisRecord(record, shardID, streamARN); err != nil {
-				a.logger.Errorw("Failed to send CloudEvent", zap.Error(err))
-				errs = append(errs, err)
-			}
-		}
+		records = append(records, recordsOutput.Records...)
 
 		// remove old input
 		inputs = append(inputs[:i], inputs[i+1:]...)
@@ -177,31 +182,18 @@ func (a *adapter) processInputs(inputs []kinesis.GetRecordsInput, shardIDs []*st
 		inputs = append(inputs, input)
 	}
 
-	return utilerrors.NewAggregate(errs)
+	return records, utilerrors.NewAggregate(errs)
 }
 
-func (a *adapter) sendKinesisRecord(record *kinesis.Record, shardID, streamARN *string) error {
+func (a *adapter) sendKinesisRecord(record *kinesis.Record) error {
 	a.logger.Infof("Processing record ID: %s", *record.SequenceNumber)
 
-	data := &Event{
-		EventName:      aws.String("aws:kinesis:record"),
-		EventSourceARN: streamARN,
-		EventSource:    aws.String("aws:kinesis"),
-		AWSRegion:      &a.arn.Region,
-		Kinesis: Kinesis{
-			PartitionKey:         record.PartitionKey,
-			Data:                 record.Data,
-			SequenceNumber:       record.SequenceNumber,
-			KinesisSchemaVersion: aws.String("1.0"),
-		},
-	}
-
 	event := cloudevents.NewEvent(cloudevents.VersionV1)
 	event.SetType(v1alpha1.AWSEventType(a.arn.Service, v1alpha1.AWSKinesisGenericEventType))
 	event.SetSubject(*record.PartitionKey)
 	event.SetSource(a.arn.String())
-	event.SetID(fmt.Sprintf("%s:%s", *shardID, *record.SequenceNumber))
-	if err := event.SetData(cloudevents.ApplicationJSON, data); err != nil {
+	event.SetID(*record.SequenceNumber)
+	if err := event.SetData(cloudevents.ApplicationJSON, record); err != nil {
 		return fmt.Errorf("failed to set event data: %w", err)
 	}
 
diff --git a/pkg/adapter/awskinesissource/events.go b/pkg/adapter/awskinesissource/events.go
deleted file mode 100644
index 2d7f8ec..0000000
--- a/pkg/adapter/awskinesissource/events.go
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
-Copyright (c) 2019-2020 TriggerMesh Inc.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-   http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package awskinesissource
-
-// Event represents an Amazon Kinesis Data Streams event.
-type Event struct {
-	EventID        *string `json:"eventID"`
-	EventVersion   *string `json:"eventVersion"`
-	Kinesis        Kinesis `json:"kinesis"`
-	EventName      *string `json:"eventName"`
-	EventSourceARN *string `json:"eventSourceARN"`
-	EventSource    *string `json:"eventSource"`
-	AWSRegion      *string `json:"awsRegion"`
-}
-
-// Kinesis represents Kinesis stream item structure
-type Kinesis struct {
-	PartitionKey         *string `json:"partitionKey"`
-	Data                 []byte  `json:"data"`
-	SequenceNumber       *string `json:"sequenceNumber"`
-	KinesisSchemaVersion *string `json:"kinesisSchemaVersion"`
-}
diff --git a/pkg/adapter/awssqssource/adapter.go b/pkg/adapter/awssqssource/adapter.go
index fdc5016..e07adbf 100644
--- a/pkg/adapter/awssqssource/adapter.go
+++ b/pkg/adapter/awssqssource/adapter.go
@@ -19,7 +19,6 @@ package awssqssource
 import (
 	"context"
 	"fmt"
-	"time"
 
 	"go.uber.org/zap"
 
@@ -84,56 +83,47 @@ func NewAdapter(ctx context.Context, envAcc pkgadapter.EnvConfigAccessor, ceClie
 	}
 }
 
-const waitTimeoutSec = 20
-
 // Start implements adapter.Adapter.
 func (a *adapter) Start(stopCh <-chan struct{}) error {
 	url, err := a.queueLookup(a.arn.Resource)
 	if err != nil {
-		a.logger.Fatalw("Unable to find URL of SQS queue "+a.arn.Resource, zap.Error(err))
+		a.logger.Errorw("Unable to find URL of SQS queue "+a.arn.Resource, zap.Error(err))
+		return err
 	}
 
 	queueURL := *url.QueueUrl
 	a.logger.Infof("Listening to SQS queue at URL: %s", queueURL)
 
-	// Look for new messages every 5 seconds
-	ticker := time.NewTicker(5 * time.Second)
-	for range ticker.C {
-		msgs, err := a.getMessages(queueURL, waitTimeoutSec)
-		if err != nil {
-			a.logger.Errorw("Failed to get messages from SQS queue", "error", err)
-			continue
-		}
-
-		// Only push if there are messages on the queue
-		if len(msgs) < 1 {
-			continue
-		}
+	backoff := common.NewBackoff()
 
-		attributes, err := a.sqsClient.GetQueueAttributes(&sqs.GetQueueAttributesInput{
-			AttributeNames: []*string{aws.String("QueueArn")},
-			QueueUrl:       url.QueueUrl,
-		})
+	err = backoff.Run(stopCh, func(ctx context.Context) (bool, error) {
+		resetBackoff := false
+		messages, err := a.getMessages(queueURL)
 		if err != nil {
-			a.logger.Errorw("Failed to get queue attributes", "error", err)
-			continue
+			a.logger.Errorw("Failed to get messages from SQS queue", "error", err)
+			return resetBackoff, nil
 		}
 
-		err = a.sendSQSEvent(msgs[0], attributes.Attributes["QueueArn"])
-		if err != nil {
-			a.logger.Errorw("Failed to send event", "error", err)
-			continue
+		for _, message := range messages {
+			err = a.sendSQSEvent(message)
+			if err != nil {
+				a.logger.Errorw("Failed to send event", "error", err)
+				continue
+			}
+
+			// Delete message from queue if we pushed successfully
+			err = a.deleteMessage(queueURL, message.ReceiptHandle)
+			if err != nil {
+				a.logger.Errorw("Failed to delete message from SQS queue", "error", err)
+				continue
+			}
+			resetBackoff = true
 		}
 
-		// Delete message from queue if we pushed successfully
-		err = a.deleteMessage(queueURL, msgs[0].ReceiptHandle)
-		if err != nil {
-			a.logger.Errorw("Failed to delete message from SQS queue", "error", err)
-			continue
-		}
-	}
+		return resetBackoff, nil
+	})
 
-	return nil
+	return err
 }
 
 // queueLookup finds the URL for a given queue name in the user's env.
@@ -146,14 +136,11 @@ func (a *adapter) queueLookup(queueName string) (*sqs.GetQueueUrlOutput, error)
 
 // getMessages returns the parsed messages from SQS if any. If an error
 // occurs that error will be returned.
-func (a *adapter) getMessages(queueURL string, waitTimeout int64) ([]*sqs.Message, error) {
+func (a *adapter) getMessages(queueURL string) ([]*sqs.Message, error) {
 	params := sqs.ReceiveMessageInput{
-		AttributeNames: aws.StringSlice([]string{"All"}),
+		AttributeNames: aws.StringSlice([]string{sqs.QueueAttributeNameAll}),
 		QueueUrl:       &queueURL,
 	}
-	if waitTimeout > 0 {
-		params.WaitTimeSeconds = &waitTimeout
-	}
 	resp, err := a.sqsClient.ReceiveMessage(&params)
 	if err != nil {
 		return nil, err
@@ -161,27 +148,20 @@ func (a *adapter) getMessages(queueURL string, waitTimeout int64) ([]*sqs.Messag
 	return resp.Messages, nil
 }
 
-func (a *adapter) sendSQSEvent(msg *sqs.Message, queueARN *string) error {
+func (a *adapter) sendSQSEvent(msg *sqs.Message) error {
 	a.logger.Infof("Processing message with ID: %s", *msg.MessageId)
 
-	data := &Event{
-		MessageID:         msg.MessageId,
-		ReceiptHandle:     msg.ReceiptHandle,
-		Body:              msg.Body,
-		Attributes:        msg.Attributes,
-		MessageAttributes: msg.MessageAttributes,
-		Md5OfBody:         msg.MD5OfBody,
-		EventSource:       aws.String("aws:sqs"),
-		EventSourceARN:    queueARN,
-		AwsRegion:         &a.arn.Region,
+	// TODO: work on CE attributes contract
+	subject, exist := msg.Attributes[sqs.MessageSystemAttributeNameSenderId]
+	if !exist {
+		subject = msg.MessageId
 	}
-
 	event := cloudevents.NewEvent(cloudevents.VersionV1)
 	event.SetType(v1alpha1.AWSEventType(a.arn.Service, v1alpha1.AWSSQSGenericEventType))
-	event.SetSubject(*msg.MessageId)
+	event.SetSubject(*subject)
 	event.SetSource(a.arn.String())
 	event.SetID(*msg.MessageId)
-	if err := event.SetData(cloudevents.ApplicationJSON, data); err != nil {
+	if err := event.SetData(cloudevents.ApplicationJSON, msg); err != nil {
 		return fmt.Errorf("failed to set event data: %w", err)
 	}
 
diff --git a/pkg/adapter/awssqssource/events.go b/pkg/adapter/awssqssource/events.go
deleted file mode 100644
index 1042484..0000000
--- a/pkg/adapter/awssqssource/events.go
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
-Copyright (c) 2019-2020 TriggerMesh Inc.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-   http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package awssqssource
-
-import (
-	"github.com/aws/aws-sdk-go/service/sqs"
-)
-
-// Event represents the payload of an AWS SQS Event.
-type Event struct {
-	MessageID         *string                               `json:"messageId"`
-	ReceiptHandle     *string                               `json:"receiptHandle"`
-	Body              *string                               `json:"body"`
-	Attributes        map[string]*string                    `json:"attributes"`
-	MessageAttributes map[string]*sqs.MessageAttributeValue `json:"messageAttributes"`
-	Md5OfBody         *string                               `json:"md5OfBody"`
-	EventSource       *string                               `json:"eventSource"`
-	EventSourceARN    *string                               `json:"eventSourceARN"`
-	AwsRegion         *string                               `json:"awsRegion"`
-}
diff --git a/pkg/adapter/common/backoff.go b/pkg/adapter/common/backoff.go
new file mode 100644
index 0000000..576be87
--- /dev/null
+++ b/pkg/adapter/common/backoff.go
@@ -0,0 +1,116 @@
+/*
+Copyright (c) 2020 TriggerMesh Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package common
+
+import (
+	"context"
+	"math"
+	"time"
+)
+
+// default values for backoff
+const (
+	expFactor = 2
+
+	minBackoff = 1 * time.Second
+	maxBackoff = 32 * time.Second
+)
+
+// Backoff provides a simple exponential backoff mechanism
+type Backoff struct {
+	step     int
+	factor   float64
+	min, max time.Duration
+}
+
+// RunFunc is a user function that polls data from a source and sends it as
+// a cloudevent to a sink.
+// RunFunc must return (bool, error) values where bool is true if poll backoff duration
+// must be reset and error is the result of function execution.
+type RunFunc func(context.Context) (bool, error)
+
+// NewBackoff accepts optional values for minimum and maximum wait period
+// and return new instance of Backoff structure
+func NewBackoff(args ...time.Duration) *Backoff {
+	backoff := &Backoff{
+		step:   0,
+		factor: expFactor,
+		min:    minBackoff,
+		max:    maxBackoff,
+	}
+
+	switch len(args) {
+	case 1:
+		if args[0] <= backoff.max {
+			backoff.min = args[0]
+		}
+	case 2:
+		if args[0] <= args[1] {
+			backoff.min = args[0]
+			backoff.max = args[1]
+		}
+	}
+
+	return backoff
+}
+
+// Duration can be used to get exponential backoff duration calculated for each new step
+func (b *Backoff) Duration() time.Duration {
+	dur := time.Duration(float64(b.min)*math.Pow(b.factor, float64(b.step)) - float64(1*time.Second))
+
+	switch {
+	case dur < b.min:
+		b.step++
+		return b.min
+	case dur > b.max:
+		return b.max
+	default:
+		b.step++
+		return dur
+	}
+}
+
+// Run is a blocking function that executes RunFunc until stopCh receives the value
+// or function returns an error
+func (b *Backoff) Run(stopCh <-chan struct{}, fn RunFunc) error {
+	timer := time.NewTimer(0)
+	defer timer.Stop()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	for {
+		select {
+		case <-stopCh:
+			return nil
+		case <-timer.C:
+			reset, err := fn(ctx)
+			if err != nil {
+				return err
+			}
+			if reset {
+				b.Reset()
+			}
+			timer.Reset(b.Duration())
+		}
+	}
+}
+
+// Reset sets step counter to zero.
+func (b *Backoff) Reset() {
+	b.step = 0
+}


Test patch:
diff --git a/pkg/adapter/awscodecommitsource/adapter_test.go b/pkg/adapter/awscodecommitsource/adapter_test.go
index 682865a..746a7e5 100644
--- a/pkg/adapter/awscodecommitsource/adapter_test.go
+++ b/pkg/adapter/awscodecommitsource/adapter_test.go
@@ -79,9 +79,10 @@ func TestSendPREvent(t *testing.T) {
 	gotEvents := ceClient.Sent()
 	assert.Len(t, gotEvents, 1, "Expected 1 event, got %d", len(gotEvents))
 
-	wantData := `{"pullRequest":{"ApprovalRules":null,"AuthorArn":null,"ClientRequestToken":null,"CreationDate":null,"Description":null,"LastActivityDate":null,"PullRequestId":"12345","PullRequestStatus":null,"PullRequestTargets":null,"RevisionId":null,"Title":null},"eventType":null,"repository":"","branch":"","eventSource":"aws:codecommit","awsRegion":""}`
-	gotData := string(gotEvents[0].Data())
-	assert.EqualValues(t, wantData, gotData, "Expected event %q, got %q", wantData, gotData)
+	var gotData codecommit.PullRequest
+	err = gotEvents[0].DataAs(&gotData)
+	assert.NoError(t, err)
+	assert.EqualValues(t, *pr, gotData, "Expected event %q, got %q", *pr, gotData)
 }
 
 func TestSendPushEvent(t *testing.T) {
@@ -101,9 +102,10 @@ func TestSendPushEvent(t *testing.T) {
 	gotEvents := ceClient.Sent()
 	assert.Len(t, gotEvents, 1, "Expected 1 event, got %d", len(gotEvents))
 
-	wantData := `{"commit":{"AdditionalData":null,"Author":null,"CommitId":"12345","Committer":null,"Message":null,"Parents":null,"TreeId":null},"commitRepository":"","commitBranch":"","commitHash":null,"eventSource":"aws:codecommit","awsRegion":""}`
-	gotData := string(gotEvents[0].Data())
-	assert.EqualValues(t, wantData, gotData, "Expected event %q, got %q", wantData, gotData)
+	var gotData codecommit.Commit
+	err = gotEvents[0].DataAs(&gotData)
+	assert.NoError(t, err)
+	assert.EqualValues(t, *commit, gotData, "Expected event %q, got %q", *commit, gotData)
 }
 
 func TestProcessCommits(t *testing.T) {
diff --git a/pkg/adapter/awsdynamodbsource/adapter_test.go b/pkg/adapter/awsdynamodbsource/adapter_test.go
index 82b3bbf..8a06264 100644
--- a/pkg/adapter/awsdynamodbsource/adapter_test.go
+++ b/pkg/adapter/awsdynamodbsource/adapter_test.go
@@ -169,7 +169,7 @@ func TestGetLatestRecords(t *testing.T) {
 		getRecordsOutputError: errors.New("fake getrecords error"),
 	}
 
-	err := a.processLatestRecords(shardIterator)
+	_, err := a.processLatestRecords(shardIterator)
 	assert.Error(t, err)
 
 	a.dyndbClient = mockedDynamoStreamsClient{
@@ -183,7 +183,7 @@ func TestGetLatestRecords(t *testing.T) {
 		getRecordsOutputError: nil,
 	}
 
-	err = a.processLatestRecords(shardIterator)
+	_, err = a.processLatestRecords(shardIterator)
 	assert.NoError(t, err)
 }
 
diff --git a/pkg/adapter/awskinesissource/adapter_test.go b/pkg/adapter/awskinesissource/adapter_test.go
index 25e9707..19bf551 100644
--- a/pkg/adapter/awskinesissource/adapter_test.go
+++ b/pkg/adapter/awskinesissource/adapter_test.go
@@ -52,8 +52,6 @@ func (m mockedGetShardIterator) GetShardIterator(in *kinesis.GetShardIteratorInp
 }
 
 func TestProcessInputs(t *testing.T) {
-	streamARN := aws.String("testARN")
-
 	now := time.Now()
 	records := []*kinesis.Record{
 		{
@@ -67,6 +65,7 @@ func TestProcessInputs(t *testing.T) {
 	a := &adapter{
 		logger:   loggingtesting.TestLogger(t),
 		ceClient: adaptertest.NewTestClient(),
+		stream:   "arn:aws:kinesis:us-east-1:123456789012:stream/foo",
 	}
 
 	a.knsClient = mockedGetRecords{
@@ -81,7 +80,7 @@ func TestProcessInputs(t *testing.T) {
 		{},
 	}
 
-	err := a.processInputs(inputs, []*string{aws.String("shardID")}, streamARN)
+	_, err := a.processInputs(inputs)
 	assert.NoError(t, err)
 
 	const errMsg = "fake error"
@@ -91,7 +90,7 @@ func TestProcessInputs(t *testing.T) {
 		err:  errors.New(errMsg),
 	}
 
-	err = a.processInputs(inputs, []*string{aws.String("shardID")}, streamARN)
+	_, err = a.processInputs(inputs)
 	assert.EqualError(t, err, errMsg)
 }
 
@@ -109,7 +108,7 @@ func TestGetRecordsInputs(t *testing.T) {
 		{ShardId: aws.String("1")},
 	}
 
-	inputs, _ := a.getRecordsInputs(shards)
+	inputs := a.getRecordsInputs(shards)
 	assert.Equal(t, 1, len(inputs))
 
 	a.knsClient = mockedGetShardIterator{
@@ -117,13 +116,11 @@ func TestGetRecordsInputs(t *testing.T) {
 		err:  errors.New("fake error"),
 	}
 
-	inputs, _ = a.getRecordsInputs(shards)
+	inputs = a.getRecordsInputs(shards)
 	assert.Equal(t, 0, len(inputs))
 }
 
 func TestSendCloudevent(t *testing.T) {
-	streamARN := aws.String("testARN")
-
 	ceClient := adaptertest.NewTestClient()
 
 	a := &adapter{
@@ -138,13 +135,14 @@ func TestSendCloudevent(t *testing.T) {
 		PartitionKey:   aws.String("key"),
 	}
 
-	err := a.sendKinesisRecord(&record, aws.String(""), streamARN)
+	err := a.sendKinesisRecord(&record)
 	assert.NoError(t, err)
 
 	gotEvents := ceClient.Sent()
 	assert.Len(t, gotEvents, 1, "Expected 1 event, got %d", len(gotEvents))
 
-	wantData := `{"eventID":null,"eventVersion":null,"kinesis":{"partitionKey":"key","data":"Zm9v","sequenceNumber":"1","kinesisSchemaVersion":"1.0"},"eventName":"aws:kinesis:record","eventSourceARN":"testARN","eventSource":"aws:kinesis","awsRegion":""}`
-	gotData := string(gotEvents[0].Data())
-	assert.EqualValues(t, wantData, gotData, "Expected event %q, got %q", wantData, gotData)
+	var gotData kinesis.Record
+	err = gotEvents[0].DataAs(&gotData)
+	assert.NoError(t, err)
+	assert.EqualValues(t, record, gotData, "Expected event %q, got %q", record, gotData)
 }
diff --git a/pkg/adapter/awssqssource/adapter_test.go b/pkg/adapter/awssqssource/adapter_test.go
index cb09e66..77c2515 100644
--- a/pkg/adapter/awssqssource/adapter_test.go
+++ b/pkg/adapter/awssqssource/adapter_test.go
@@ -122,7 +122,7 @@ func TestGetMessages(t *testing.T) {
 			sqsClient: mockedReceiveMsgs{Resp: c.Resp, err: c.err},
 		}
 
-		msgs, err := a.getMessages(queueURL, waitTimeoutSec)
+		msgs, err := a.getMessages(queueURL)
 		assert.Equal(t, c.err, err)
 		assert.Equal(t, len(c.Expected), len(msgs))
 	}
@@ -143,15 +143,16 @@ func TestPushMessage(t *testing.T) {
 		ceClient: ceClient,
 	}
 
-	err := a.sendSQSEvent(&msg, aws.String("testQueueARN"))
+	err := a.sendSQSEvent(&msg)
 	assert.NoError(t, err)
 
 	gotEvents := ceClient.Sent()
 	assert.Len(t, gotEvents, 1, "Expected 1 event, got %d", len(gotEvents))
 
-	wantData := `{"messageId":"foo","receiptHandle":null,"body":"bar","attributes":{"SentTimestamp":"1549540781"},"messageAttributes":null,"md5OfBody":null,"eventSource":"aws:sqs","eventSourceARN":"testQueueARN","awsRegion":""}`
-	gotData := string(gotEvents[0].Data())
-	assert.EqualValues(t, wantData, gotData, "Expected event %q, got %q", wantData, gotData)
+	var gotData sqs.Message
+	err = gotEvents[0].DataAs(&gotData)
+	assert.NoError(t, err)
+	assert.EqualValues(t, msg, gotData, "Expected event %q, got %q", msg, gotData)
 }
 
 func TestDeleteMessage(t *testing.T) {
diff --git a/pkg/adapter/common/backoff_test.go b/pkg/adapter/common/backoff_test.go
new file mode 100644
index 0000000..2bfa18e
--- /dev/null
+++ b/pkg/adapter/common/backoff_test.go
@@ -0,0 +1,173 @@
+/*
+Copyright (c) 2020 TriggerMesh Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package common
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewBackoff(t *testing.T) {
+	testCases := map[string]struct {
+		wantMin, wantMax time.Duration
+		gotMin, gotMax   time.Duration
+	}{
+		"defaults": {
+			gotMin: minBackoff,
+			gotMax: maxBackoff,
+		},
+		"all correct": {
+			wantMin: time.Second,
+			gotMin:  time.Second,
+			wantMax: time.Minute,
+			gotMax:  time.Minute,
+		},
+		"min > default max": {
+			wantMin: time.Hour,
+			gotMin:  minBackoff,
+			gotMax:  maxBackoff,
+		},
+		"min > max": {
+			wantMin: time.Hour,
+			gotMin:  minBackoff,
+			wantMax: time.Minute,
+			gotMax:  maxBackoff,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			bo := NewBackoff()
+			switch {
+			case tc.wantMin != 0 && tc.wantMax != 0:
+				bo = NewBackoff(tc.wantMin, tc.wantMax)
+			case tc.wantMin != 0:
+				bo = NewBackoff(tc.wantMin)
+			}
+			assert.Equal(t, bo.min, tc.gotMin, "backoff min duration has unexpected value")
+			assert.Equal(t, bo.max, tc.gotMax, "backoff max duration has unexpected value")
+		})
+	}
+}
+
+func TestDuration(t *testing.T) {
+	testCases := map[string]struct {
+		step              int
+		min, max, wantDur time.Duration
+	}{
+		"first step": {
+			step:    0,
+			min:     time.Second,
+			max:     time.Minute,
+			wantDur: time.Second,
+		},
+		"third step": {
+			step:    2,
+			min:     time.Second,
+			wantDur: 3 * time.Second,
+		},
+		"tenth step": {
+			step:    9,
+			min:     time.Second,
+			max:     time.Minute,
+			wantDur: time.Minute,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			bo := NewBackoff()
+			switch {
+			case tc.min != 0 && tc.max != 0:
+				bo = NewBackoff(tc.min, tc.max)
+			case tc.min != 0:
+				bo = NewBackoff(tc.min)
+			}
+
+			bo.step = tc.step
+			assert.Equal(t, tc.wantDur, bo.Duration())
+		})
+	}
+}
+
+func TestRun(t *testing.T) {
+	var counter int
+	start := time.Now()
+
+	testCases := map[string]struct {
+		fn         func(context.Context) (bool, error)
+		waitReturn bool
+		err        error
+	}{
+		"force stop with no errs": {
+			waitReturn: false,
+		},
+		"fn returns err": {
+			fn: func(ctx context.Context) (bool, error) {
+				return false, assert.AnError
+			},
+			waitReturn: true,
+			err:        assert.AnError,
+		},
+		"fn resets backoff duration": {
+			fn: func(ctx context.Context) (bool, error) {
+				counter++
+				if counter > 3 {
+					if time.Since(start).Round(time.Second) != 3*time.Second {
+						return true, fmt.Errorf("did we reset the backoff?")
+					}
+					// trigger Run() termination with expected error
+					return true, assert.AnError
+				}
+				// reset backoff duration to 1 sec for each iteration
+				return true, nil
+			},
+			waitReturn: true,
+			err:        assert.AnError,
+		},
+	}
+
+	errCh := make(chan error)
+	defer close(errCh)
+
+	for name, tc := range testCases {
+		stopCh := make(chan struct{})
+		t.Run(name, func(t *testing.T) {
+			bo := NewBackoff()
+			go func() {
+				err := bo.Run(stopCh, tc.fn)
+				errCh <- err
+			}()
+			if tc.waitReturn {
+				defer close(stopCh)
+			} else {
+				close(stopCh)
+			}
+
+			err := <-errCh
+			if tc.err != nil {
+				assert.EqualError(t, err, tc.err.Error())
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}