luciano/vmware-go-kcl-v2
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

KCL: remove panic in shard consumer

Browse source 
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
This commit is contained in:
Tao Jiang 2018-05-31 05:26:49 -07:00
parent 3120d89ae8
commit 48fd4dd51c
1 changed files with 7 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
clientlibrary/worker/shard-consumer.go
View file

@ -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