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

More properly handle errors from KCL

Browse source
This commit is contained in:
Xavi Ramirez 2017-08-18 01:42:36 +00:00
parent 4b4af77ba2
commit b385278e04
1 changed files with 8 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
kcl/kcl.go
View file

@ -140,6 +140,7 @@ func New(
recordProcessor: recordProcessor, recordProcessor: recordProcessor,
nextCheckpointPair: SequencePair{}, nextCheckpointPair: SequencePair{},
wasAskedToShutdown: false,
} }
} }
@ -150,6 +151,7 @@ type KCLProcess struct {
recordProcessor RecordProcessor recordProcessor RecordProcessor
nextCheckpointPair SequencePair nextCheckpointPair SequencePair
wasAskedToShutdown bool
} }
func (kclp *KCLProcess) Checkpoint(pair SequencePair) { func (kclp *KCLProcess) Checkpoint(pair SequencePair) {
@ -163,7 +165,7 @@ func (kclp *KCLProcess) Checkpoint(pair SequencePair) {
func (kclp *KCLProcess) Shutdown() { func (kclp *KCLProcess) Shutdown() {
kclp.ioHandler.writeError("Checkpoint shutdown") kclp.ioHandler.writeError("Checkpoint shutdown")
kclp.sendCheckpoint(nil, nil) // nil sequence num is signal to shutdown kclp.wasAskedToShutdown = true
} }
func (kclp *KCLProcess) handleCheckpointAction(action ActionCheckpoint) error { func (kclp *KCLProcess) handleCheckpointAction(action ActionCheckpoint) error {
@ -275,7 +277,6 @@ func (kclp *KCLProcess) Run() {
err = kclp.handleLine(line) err = kclp.handleLine(line)
if err != nil { if err != nil {
kclp.ioHandler.writeError(fmt.Sprintf("ERR Handle line: %+#v", err)) kclp.ioHandler.writeError(fmt.Sprintf("ERR Handle line: %+#v", err))
return
} }
kclp.ckpmux.Lock() kclp.ckpmux.Lock()
@ -289,6 +290,11 @@ func (kclp *KCLProcess) Run() {
} else { } else {
kclp.nextCheckpointPair = SequencePair{} kclp.nextCheckpointPair = SequencePair{}
} }
} else if kclp.wasAskedToShutdown {
err := kclp.sendCheckpoint(nil, nil) // nil sequence num is signal to shutdown
if err != nil {
kclp.ioHandler.writeError(fmt.Sprintf("ERR shutdown ask: %+#v", err))
}
} }
kclp.ckpmux.Unlock() kclp.ckpmux.Unlock()
} }