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

Merge pull request #33 from wwwjfy/fix-shutdown-requested

Browse source 
Fix shutdown requested
This commit is contained in:
Xavi 2018-12-11 16:28:37 -08:00 committed by GitHub
commit 23c68b42ff
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 9 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
cmd/consumer/main.go
View file

@ -55,6 +55,9 @@ func (srp *sampleRecordProcessor) Shutdown(reason string) error {
if reason == "TERMINATE" {
fmt.Fprintf(os.Stderr, "Was told to terminate, will attempt to checkpoint.\n")
srp.checkpointer.Shutdown()
} else if reason == "SHUTDOWN_REQUESTED" {
fmt.Fprintf(os.Stderr, "Got shutdown requested, attempt to checkpoint.\n")
srp.checkpointer.Shutdown()
} else {
fmt.Fprintf(os.Stderr, "Shutting down due to failover. Will not checkpoint.\n")
}
@ -62,11 +65,6 @@ func (srp *sampleRecordProcessor) Shutdown(reason string) error {
}
func main() {
f, err := os.Create("/tmp/kcl_stderr")
if err != nil {
panic(err)
}
defer f.Close()
kclProcess := kcl.New(os.Stdin, os.Stdout, os.Stderr, newSampleRecordProcessor())
kclProcess.Run()
}

7
kcl/kcl.go
View file

@ -238,8 +238,13 @@ func (kclp *KCLProcess) handleLine(line string) (string, error) {
case ActionShutdown:
kclp.ioHandler.writeError("Received shutdown action...")
reason := action.Reason
// Shutdown should block until it's safe to shutdown the process
err = kclp.recordProcessor.Shutdown(action.Reason)
if action.Action == "shutdownRequested" {
reason = "SHUTDOWN_REQUESTED"
}
err = kclp.recordProcessor.Shutdown(reason)
if err != nil { // Log error and continue shutting down
kclp.ioHandler.writeError(fmt.Sprintf("ERR shutdown: %+#v", err))
}