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

KCL: Update shard sync to remove not existed shard

Browse source 
Need to remove shard not longer existed in Kinesis from shardStatus
cache.

Change-Id: I09b4a4c3c6480b8300fa937e6073dcd578156b29
This commit is contained in:
Tao Jiang 2018-04-21 19:58:51 -07:00
parent e1071abc80
commit 5ef4338a22
1 changed files with 23 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
src/clientlibrary/worker/worker.go
View file

@ -168,12 +168,6 @@ func (w *Worker) initialize() error {
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
w.waitGroup = &wg w.waitGroup = &wg
err = w.getShardIDs("")
if err != nil {
log.Errorf("Error getting Kinesis shards: %s", err)
return err
}
log.Info("Initialization complete.") log.Info("Initialization complete.")
return nil return nil
@ -199,12 +193,13 @@ func (w *Worker) newShardConsumer(shard *shardStatus) *ShardConsumer {
// eventLoop // eventLoop
func (w *Worker) eventLoop() { func (w *Worker) eventLoop() {
for { for {
err := w.getShardIDs("") err := w.syncShard()
if err != nil { if err != nil {
log.Errorf("Error getting Kinesis shards: %v", err) log.Errorf("Error getting Kinesis shards: %v", err)
// Back-off? // Back-off?
time.Sleep(500 * time.Millisecond) time.Sleep(500 * time.Millisecond)
} }
log.Infof("Found %d shards", len(w.shardStatus)) log.Infof("Found %d shards", len(w.shardStatus))
// Count the number of leases hold by this worker // Count the number of leases hold by this worker
@ -271,7 +266,9 @@ func (w *Worker) eventLoop() {
} }
// List all ACTIVE shard and store them into shardStatus table // List all ACTIVE shard and store them into shardStatus table
func (w *Worker) getShardIDs(startShardID string) error { // If shard has been removed, need to exclude it from cached shard status.
func (w *Worker) getShardIDs(startShardID string, shardInfo map[string]bool) error {
// The default pagination limit is 100.
args := &kinesis.DescribeStreamInput{ args := &kinesis.DescribeStreamInput{
StreamName: aws.String(w.streamName), StreamName: aws.String(w.streamName),
} }
@ -289,6 +286,8 @@ func (w *Worker) getShardIDs(startShardID string) error {
var lastShardID string var lastShardID string
for _, s := range streamDesc.StreamDescription.Shards { for _, s := range streamDesc.StreamDescription.Shards {
// record avail shardId from fresh reading from Kinesis
shardInfo[*s.ShardId] = true
// found new shard // found new shard
if _, ok := w.shardStatus[*s.ShardId]; !ok { if _, ok := w.shardStatus[*s.ShardId]; !ok {
log.Debugf("Found shard with id %s", *s.ShardId) log.Debugf("Found shard with id %s", *s.ShardId)
@ -304,7 +303,7 @@ func (w *Worker) getShardIDs(startShardID string) error {
} }
if *streamDesc.StreamDescription.HasMoreShards { if *streamDesc.StreamDescription.HasMoreShards {
err := w.getShardIDs(lastShardID) err := w.getShardIDs(lastShardID, shardInfo)
if err != nil { if err != nil {
return err return err
} }
@ -312,3 +311,18 @@ func (w *Worker) getShardIDs(startShardID string) error {
return nil return nil
} }
// syncShard to sync the cached shard info with actual shard info from Kinesis
func (w *Worker) syncShard() error {
shardInfo := make(map[string]bool)
err := w.getShardIDs("", shardInfo)
for _, shard := range w.shardStatus {
// The cached shard no longer existed, remove it.
if _, ok := shardInfo[shard.ID]; !ok {
delete(w.shardStatus, shard.ID)
}
}
return err
}