Method ProcessRecords

input/kcl.go:306–350

ProcessRecords process data records. vmware kcl will invoke this method to deliver data records. Upon fail over, the new instance will get records with sequence number greater than checkpoint position for each partition key. 'input' provides the records to be processed as well as information and c

(input *interfaces.ProcessRecordsInput)

// 'input' provides the records to be processed as well as information and
// capabilities related to them (eg checkpointing).
func (p *recordProcessor) ProcessRecords(input *interfaces.ProcessRecordsInput) {
	// Control read throughput to prevent throttling.
	//
	// Kinesis imposes limits on GetRecords, see
	// https://docs.aws.amazon.com/streams/latest/dev/service-sizes-and-limits.html
	//
	// Each shard can support up to a maximum total data read rate of 2 MiB
	// per second via GetRecords. If a call to GetRecords returns 10 MiB, the
	// maximum size GetRecords is allowed to return, subsequent calls made
	// within the next 5 seconds will meet a ProvisionedThroughputExceededException.
	//
	// Limiting the number of records per call would work but would increase
	// the number of performed IO syscalls and will increase the risk to meet
	// the limits imposed by AWS on API calls.
	//
	// The strategy we're using is to not limit MaxRecords but sleeping for 6s.
	// Doing so, we're guaranteed to never exceed the per-shard read througput
	// limit of 2MB/s, while being close to it on data peaks. This has the
	// added advantage of reducing the number of IO syscalls.
	time.Sleep(6 * time.Second)

	// Skip if no records
	if len(input.Records) == 0 {
		log.Debug("No records to process")
		return
	}

	// Send the records to Baker pipeline
	var nlines int64
	for _, v := range input.Records {
		nlines += int64(bytes.Count(v.Data, []byte{'\n'}))
		p.inch <- &baker.Data{Bytes: v.Data}
	}

	// Increment the total number of lines processed by the KCL worker.
	// note: p.nlines is shared among all record processors
	atomic.AddInt64(p.nlines, nlines)

	// Checkpoint it after processing this batch
	lastRecordSequenceNumber := input.Records[len(input.Records)-1].SequenceNumber
	log.Debugf("Processed %d records: checkpoint=%s, msBehindLatest=%v", len(input.Records), aws.StringValue(lastRecordSequenceNumber), input.MillisBehindLatest)
	if err := input.Checkpointer.Checkpoint(lastRecordSequenceNumber); err != nil {
		log.Errorf("Error checkpointing at %s: %s", *lastRecordSequenceNumber, err)
	}
}

// kclMetrics implements kcl metrics.MonitoringService.
type kclDatadogMetrics struct {