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

Added shutdown method to *Checkerpointer struct

Browse source
This commit is contained in:
Xavi Ramirez 2017-05-22 23:06:56 +00:00
parent 5f507ab116
commit 458e66e321
2 changed files with 5 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
cmd/consumer/main.go
View file

@ -59,7 +59,7 @@ func (srp *SampleRecordProcessor) ProcessRecords(records []kcl.Record) error {
func (srp *SampleRecordProcessor) Shutdown(reason string) error { func (srp *SampleRecordProcessor) Shutdown(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(nil, nil) srp.checkpointer.Shutdown()
} 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")
} }

4
kcl/kcl.go
View file

@ -108,6 +108,10 @@ func (c *Checkpointer) CheckpointWithRetry(
return nil return nil
} }
func (c *Checkpointer) Shutdown() {
c.CheckpointWithRetry(nil, nil, 5)
}
type ioHandler struct { type ioHandler struct {
inputFile io.Reader inputFile io.Reader
outputFile io.Writer outputFile io.Writer