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

checkpoint after filtered messages in the pipeline

Browse source 
checkpoint after filtered messages in the pipeline

checkpoint after filtered messages in the pipeline
This commit is contained in:
dan 2015-04-07 11:19:10 -07:00
parent 0dce2a6045
commit 2ff748a2d4
2 changed files with 16 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
pipeline.go
View file

@ -16,12 +16,13 @@ import (
// interface. It has a data type (Model) as Records come in as a byte[] and are transformed to a Model. // interface. It has a data type (Model) as Records come in as a byte[] and are transformed to a Model.
// Then they are buffered in Model form and when the buffer is full, Models's are passed to the emitter. // Then they are buffered in Model form and when the buffer is full, Models's are passed to the emitter.
type Pipeline struct { type Pipeline struct {
Buffer Buffer Buffer Buffer
Checkpoint Checkpoint Checkpoint Checkpoint
Emitter Emitter Emitter Emitter
Filter Filter Filter Filter
StreamName string StreamName string
Transformer Transformer Transformer Transformer
CheckpointFilteredRecords bool
} }
var pipelineRecoverableErrorCodes = map[string]bool{ var pipelineRecoverableErrorCodes = map[string]bool{
@ -110,6 +111,8 @@ func (p Pipeline) ProcessShard(ksis *kinesis.Kinesis, shardID string) {
if p.Filter.KeepRecord(r) { if p.Filter.KeepRecord(r) {
p.Buffer.ProcessRecord(r, v.SequenceNumber) p.Buffer.ProcessRecord(r, v.SequenceNumber)
} else if p.CheckpointFilteredRecords {
p.Buffer.ProcessRecord(nil, v.SequenceNumber)
} }
} }
} else if recordSet.NextShardIterator == "" || shardIterator == recordSet.NextShardIterator || err != nil { } else if recordSet.NextShardIterator == "" || shardIterator == recordSet.NextShardIterator || err != nil {
@ -120,7 +123,9 @@ func (p Pipeline) ProcessShard(ksis *kinesis.Kinesis, shardID string) {
} }
if p.Buffer.ShouldFlush() { if p.Buffer.ShouldFlush() {
p.Emitter.Emit(p.Buffer, p.Transformer) if p.Buffer.NumRecordsInBuffer() > 0 {
p.Emitter.Emit(p.Buffer, p.Transformer)
}
p.Checkpoint.SetCheckpoint(shardID, p.Buffer.LastSequenceNumber()) p.Checkpoint.SetCheckpoint(shardID, p.Buffer.LastSequenceNumber())
p.Buffer.Flush() p.Buffer.Flush()
} }

6
record_buffer.go
View file

@ -20,7 +20,9 @@ func (b *RecordBuffer) ProcessRecord(record interface{}, sequenceNumber string)
b.lastSequenceNumber = sequenceNumber b.lastSequenceNumber = sequenceNumber
if !b.sequenceExists(sequenceNumber) { if !b.sequenceExists(sequenceNumber) {
b.recordsInBuffer = append(b.recordsInBuffer, record) if record != nil {
b.recordsInBuffer = append(b.recordsInBuffer, record)
}
b.sequencesInBuffer = append(b.sequencesInBuffer, sequenceNumber) b.sequencesInBuffer = append(b.sequencesInBuffer, sequenceNumber)
} }
} }
@ -32,7 +34,7 @@ func (b *RecordBuffer) Records() []interface{} {
// NumRecordsInBuffer returns the number of messages in the buffer. // NumRecordsInBuffer returns the number of messages in the buffer.
func (b RecordBuffer) NumRecordsInBuffer() int { func (b RecordBuffer) NumRecordsInBuffer() int {
return len(b.sequencesInBuffer) return len(b.recordsInBuffer)
} }
// Flush empties the buffer and resets the sequence counter. // Flush empties the buffer and resets the sequence counter.