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

fix review comments

Browse source
This commit is contained in:
Tony Wang 2018-10-18 15:22:57 +08:00
parent e7b75a20ac
commit 06838386db
No known key found for this signature in database
GPG key ID: 624D973D837983DF
2 changed files with 6 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
cmd/consumer/main.go
View file

@ -51,16 +51,13 @@ func (srp *sampleRecordProcessor) ProcessRecords(records []kcl.Record) error {
return nil return nil
} }
func (srp *sampleRecordProcessor) ShutdownRequested() error {
fmt.Fprintf(os.Stderr, "Got shutdown requested, attempt to checkpoint.\n")
srp.checkpointer.Shutdown()
return nil
}
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.checkpointer.Shutdown() srp.checkpointer.Shutdown()
} else if reason == "SHUTDOWN_REQUESTED" {
fmt.Fprintf(os.Stderr, "Got shutdown requested, attempt to checkpoint.\n")
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")
} }

7
kcl/kcl.go
View file

@ -12,7 +12,6 @@ import (
type RecordProcessor interface { type RecordProcessor interface {
Initialize(shardID string, checkpointer Checkpointer) error Initialize(shardID string, checkpointer Checkpointer) error
ProcessRecords(records []Record) error ProcessRecords(records []Record) error
ShutdownRequested() error
// Shutdown this call should block until it's safe to shutdown the process // Shutdown this call should block until it's safe to shutdown the process
Shutdown(reason string) error Shutdown(reason string) error
} }
@ -237,12 +236,12 @@ func (kclp *KCLProcess) handleLine(line string) (string, error) {
case ActionShutdown: case ActionShutdown:
kclp.ioHandler.writeError("Received shutdown action...") kclp.ioHandler.writeError("Received shutdown action...")
reason := action.Reason
// Shutdown should block until it's safe to shutdown the process // Shutdown should block until it's safe to shutdown the process
if action.Action == "shutdownRequested" { if action.Action == "shutdownRequested" {
err = kclp.recordProcessor.ShutdownRequested() reason = "SHUTDOWN_REQUESTED"
} else {
err = kclp.recordProcessor.Shutdown(action.Reason)
} }
err = kclp.recordProcessor.Shutdown(reason)
if err != nil { // Log error and continue shutting down if err != nil { // Log error and continue shutting down
kclp.ioHandler.writeError(fmt.Sprintf("ERR shutdown: %+#v", err)) kclp.ioHandler.writeError(fmt.Sprintf("ERR shutdown: %+#v", err))