Properly handle leases for slow record processing about vmware-go-kcl HOT 7 CLOSED

I think adding extra go routine complicates the process model. The other alternative is using the similar model used by SWF by reporting RecordActivityTaskHeartbeat which is similar to periodic checkpointing by just renewing lease instead.

https://docs.aws.amazon.com/amazonswf/latest/apireference/API_RecordActivityTaskHeartbeat.html

from vmware-go-kcl.

The other way is to do periodic checkpointing without adding more api which doesn't mean it has to be done after processing all records in a batch.

from vmware-go-kcl.

Thanks for the quick replies.

For my particular issue, doing periodic checkpoints would actually work (I'm doing that anyway) - but reading through the code, a checkpoint doesn't renew the lease. If we could change the behavior for a checkpoint to renew the lease, this issue would be solved for me.

Happy to implement that. Or happy to hear that I read the code wrong :).

from vmware-go-kcl.

You are correct and it is a bug:
https://github.com/vmware/vmware-go-kcl/blob/master/clientlibrary/checkpoint/dynamodb-checkpointer.go#L198

The leaseTimeout should be updated when doing checkpointing similar to GetLease()

You are more than welcome to implement it. Thx!

from vmware-go-kcl.

Looking at GetLease, and specifically https://github.com/vmware/vmware-go-kcl/blob/master/clientlibrary/checkpoint/dynamodb-checkpointer.go#L173 it looks like the checkpoint will already be set in that function. Since CheckpointSequence already assumes the checkpoint is set in the shard, I think this will work as intended with:

func (checkpointer *DynamoCheckpoint) CheckpointSequence(shard *par.ShardStatus) error {
    // We assume shard.Checkpoint has been set correctly here. Since we also want
    // to renew the lease while checkpointing in case the processing of a set of records
    // takes longer than the lease time, we simply can call GetLease here.
    return checkpointer.GetLease(shard, shard.AssignedTo)
}

If you agree, I'll put up a quick PR for this.

from vmware-go-kcl.

It should be much simpler since the caller of CheckpointSequence already has the lease. Just need to update new lease timeout.

	newLeaseTimeout := time.Now().Add(time.Duration(checkpointer.LeaseDuration) * time.Millisecond).UTC()
	newLeaseTimeoutString := newLeaseTimeout.Format(time.RFC3339)

        # update lease timeout
	shard.LeaseTimeout = newLeaseTimeout

from vmware-go-kcl.

This is actually not a bug. Kinesis document says: "Gets an Amazon Kinesis shard iterator. A shard iterator expires 5 minutes after it is returned to the requester."
https://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetShardIterator.html

which means the shard iterator will be expired even though the consumer still hold the lease. The workaround is to reduce max record number using WithMaxRecords(smaller-value).

from vmware-go-kcl.