luciano/amazon-kinesis-client-go
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #2 from Clever/shutdown-checkpoint

Browse source 
allow for empty checkpoints on shutdown
This commit is contained in:
Rafael 2017-05-17 17:52:42 -07:00 committed by GitHub
commit 8da04c944f
2 changed files with 11 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
cmd/consumer/main.go
View file

@ -31,7 +31,7 @@ func (srp *SampleRecordProcessor) Initialize(shardID string) error {
return nil return nil
} }
func (srp *SampleRecordProcessor) checkpoint(checkpointer kcl.Checkpointer, sequenceNumber string, subSequenceNumber int) { func (srp *SampleRecordProcessor) checkpoint(checkpointer kcl.Checkpointer, sequenceNumber *string, subSequenceNumber *int) {
for n := -1; n < srp.checkpointRetries; n++ { for n := -1; n < srp.checkpointRetries; n++ {
err := checkpointer.Checkpoint(sequenceNumber, subSequenceNumber) err := checkpointer.Checkpoint(sequenceNumber, subSequenceNumber)
if err == nil { if err == nil {
@ -79,7 +79,8 @@ func (srp *SampleRecordProcessor) ProcessRecords(records []kcl.Record, checkpoin
} }
} }
if time.Now().Sub(srp.lastCheckpoint) > srp.checkpointFreq { if time.Now().Sub(srp.lastCheckpoint) > srp.checkpointFreq {
srp.checkpoint(checkpointer, srp.largestSeq.String(), srp.largestSubSeq) largestSeq := srp.largestSeq.String()
srp.checkpoint(checkpointer, &largestSeq, &srp.largestSubSeq)
srp.lastCheckpoint = time.Now() srp.lastCheckpoint = time.Now()
} }
return nil return nil
@ -88,7 +89,7 @@ func (srp *SampleRecordProcessor) ProcessRecords(records []kcl.Record, checkpoin
func (srp *SampleRecordProcessor) Shutdown(checkpointer kcl.Checkpointer, reason string) error { func (srp *SampleRecordProcessor) Shutdown(checkpointer kcl.Checkpointer, reason string) error {
if reason == "TERMINATE" { if reason == "TERMINATE" {
fmt.Fprintf(os.Stderr, "Was told to terminate, will attempt to checkpoint.\n") fmt.Fprintf(os.Stderr, "Was told to terminate, will attempt to checkpoint.\n")
srp.checkpoint(checkpointer, "", 0) srp.checkpoint(checkpointer, nil, nil)
} else { } else {
fmt.Fprintf(os.Stderr, "Shutting down due to failover. Will not checkpoint.\n") fmt.Fprintf(os.Stderr, "Shutting down due to failover. Will not checkpoint.\n")
} }

12
kcl/kcl.go
View file

@ -38,7 +38,7 @@ func (ce CheckpointError) Error() string {
return ce.e return ce.e
} }
func (c Checkpointer) Checkpoint(sequenceNumber string, subSequenceNumber int) error { func (c Checkpointer) Checkpoint(sequenceNumber *string, subSequenceNumber *int) error {
c.ioHandler.writeAction(ActionCheckpoint{ c.ioHandler.writeAction(ActionCheckpoint{
Action: "checkpoint", Action: "checkpoint",
SequenceNumber: sequenceNumber, SequenceNumber: sequenceNumber,
@ -56,9 +56,9 @@ func (c Checkpointer) Checkpoint(sequenceNumber string, subSequenceNumber int) e
if !ok { if !ok {
return fmt.Errorf("expected checkpoint response, got '%s'", line.String()) return fmt.Errorf("expected checkpoint response, got '%s'", line.String())
} }
if action.Error != "" { if action.Error != nil && *action.Error != "" {
return CheckpointError{ return CheckpointError{
e: action.Error, e: *action.Error,
} }
} }
return nil return nil
@ -118,9 +118,9 @@ type ActionShutdown struct {
type ActionCheckpoint struct { type ActionCheckpoint struct {
Action string `json:"action"` Action string `json:"action"`
SequenceNumber string `json:"sequenceNumber"` SequenceNumber *string `json:"sequenceNumber,omitempty"`
SubSequenceNumber int `json:"subSequenceNumber"` SubSequenceNumber *int `json:"subSequenceNumber,omitempty"`
Error string `json:"error"` Error *string `json:"error,omitempty"`
} }
func (i ioHandler) loadAction(line string) (interface{}, error) { func (i ioHandler) loadAction(line string) (interface{}, error) {