From 48fd4dd51cd351ac8f797ff42da0b541ece73089 Mon Sep 17 00:00:00 2001 From: Tao Jiang Date: Thu, 31 May 2018 05:26:49 -0700 Subject: [PATCH] KCL: remove panic in shard consumer There might be verious reason for shard iterator to expire, such as: not enough data in shard or process even takes more than 5 minutes which cause shard iterator not refreshing enough. This change removes log.Fatal which causes panic. Panic inside go routine will bring down the whole app. Therefore, just log error and exit the go routine instead. Jira ID: CNA-1072 Change-Id: I34a8d9af7258f3ea75465e2245bbc25c2fafee35 --- clientlibrary/worker/shard-consumer.go | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/clientlibrary/worker/shard-consumer.go b/clientlibrary/worker/shard-consumer.go index 9a76309..d238943 100644 --- a/clientlibrary/worker/shard-consumer.go +++ b/clientlibrary/worker/shard-consumer.go @@ -130,7 +130,10 @@ func (sc *ShardConsumer) getRecords(shard *shardStatus) error { log.Warnf("Failed in acquiring lease on shard: %s for worker: %s", shard.ID, sc.consumerID) return nil } - log.Fatal(err) + // log and return error + log.Errorf("Error in refreshing lease on shard: %s for worker: %s. Error: %+v", + shard.ID, sc.consumerID, err) + return err } } @@ -144,14 +147,15 @@ func (sc *ShardConsumer) getRecords(shard *shardStatus) error { if err != nil { if awsErr, ok := err.(awserr.Error); ok { if awsErr.Code() == kinesis.ErrCodeProvisionedThroughputExceededException || awsErr.Code() == ErrCodeKMSThrottlingException { - log.Errorf("Error getting records from shard %v: %v", shard.ID, err) + log.Errorf("Error getting records from shard %v: %+v", shard.ID, err) retriedErrors++ // exponential backoff time.Sleep(time.Duration(2^retriedErrors*100) * time.Millisecond) continue } } - log.Fatalf("Error getting records from Kinesis that cannot be retried: %s\nRequest: %s", err, getRecordsArgs) + log.Errorf("Error getting records from Kinesis that cannot be retried: %+v\nRequest: %s", err, getRecordsArgs) + return err } retriedErrors = 0