import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor; //导入依赖的package包/类
@Override
public IRecordProcessor createProcessor() {
    return new FlumeSourceRecordProcessor(chProcessor,
                                          serializer,
                                          backOffTimeInMillis,
                                          numberRetries,
                                          checkpointIntervalMillis);
}
 

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
public IRecordProcessor createProcessor() {
	try {
		LOG.info("Creating new Managed Client Processor");
		ManagedClientProcessor p = this.managedProcessor.copy();
		createdProcessors.put(p.toString(), p);
		return p;
	} catch (Exception e) {
		LOG.error(e);
		return null;
	}
}
 

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor; //导入依赖的package包/类
@Override
public IRecordProcessor createProcessor() {
    logger.info("Creating RecordProcessor");
    if (recordProcessor == null) {
        recordProcessor = new RecordProcessor();
    }
    return recordProcessor;
}
 

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor; //导入依赖的package包/类
@Override
public IRecordProcessor createProcessor() {
    try {
        IBuffer<T> buffer = pipeline.getBuffer(configuration);
        IEmitter<U> emitter = pipeline.getEmitter(configuration);
        ITransformerBase<T, U> transformer = pipeline.getTransformer(configuration);
        IFilter<T> filter = pipeline.getFilter(configuration);
        KinesisConnectorRecordProcessor<T, U> processor =
                new KinesisConnectorRecordProcessor<T, U>(buffer, filter, emitter, transformer, configuration);
        return processor;
    } catch (Throwable t) {
        throw new RuntimeException(t);
    }
}
 

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
public IRecordProcessor createProcessor() {
    try {
        // every time we create a new processor instance, we have to embed a
        // new instance of the AggregatorGroup, to eliminate any thread
        // contention
        return new AggregatorProcessor(new AggregatorGroup(this.aggregators));
    } catch (Exception e) {
        LOG.error(e);
        return null;
    }
}
 

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor; //导入依赖的package包/类
@Override
public IRecordProcessor createProcessor() {
    return new EventRecordProcessor(receiver, workerId,
                            new KinesisCheckpointState(checkpointInterval));
}
 

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor; //导入依赖的package包/类
@Override
public IRecordProcessor createProcessor() {
    return new KafkaForwardingStreamsRecordProcessor(props, topic);
}
 

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor; //导入依赖的package包/类
@Override
public IRecordProcessor createProcessor() {
	return new Processor();
}
 

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
public IRecordProcessor createProcessor() {
    return new KinesisRecordProcessor(redisEndpoint, redisPort);
}
 

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor; //导入依赖的package包/类
@Override
public IRecordProcessor createProcessor() {
	return new StreamsRecordProcessor(cache);
}
 

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor; //导入依赖的package包/类
@Override
public IRecordProcessor createProcessor() {
	return new KinesisRawEventRecordProcessor(this.topic, this.handler, this.mapper);
}
 

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor; //导入依赖的package包/类
@Override
public IRecordProcessor createProcessor() {
    return new RecordProcessor(_buffer);
}
 

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor; //导入依赖的package包/类
@Override
public IRecordProcessor createProcessor() {
    AmazonDynamoDBClient dynamoDBClient = new AmazonDynamoDBClient(dynamoDBCredentials, new ClientConfiguration());
    dynamoDBClient.setEndpoint(dynamoDBEndpoint);
    return new StreamsRecordProcessor(dynamoDBClient, tableName);
}
 

import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public IRecordProcessor createProcessor() {
    return new TweetamoRecordProcessor();
}