func (p *KafkaPipe) Type() string {
	return "kafka"
}
Type returns Pipe type as Kafka
package pipe
import (
	"database/sql"
	"fmt"
	"golang.org/x/net/context"  
	"strings"
	"sync"
	"github.com/Shopify/sarama"
	"github.com/uber/storagetapper/config"
	"github.com/uber/storagetapper/log"
	"github.com/uber/storagetapper/types"
	"github.com/uber/storagetapper/util"
)
var (
	OffsetOldest = sarama.OffsetOldest
	OffsetNewest = sarama.OffsetNewest
)
var offsetPersistInterval int64 = 10000
var InitialOffset = OffsetNewest
var KafkaConfig *sarama.Config
type kafkaPartition struct {
	id            int32
	offset        int64
	savedOffset   int64
	consumer      sarama.PartitionConsumer
	childConsumer chan *sarama.ConsumerMessage
	nextMsg       *sarama.ConsumerMessage
}
type topicConsumer struct {
	partitions []kafkaPartition
	consumers  []chan *sarama.ConsumerMessage
	cancel     context.CancelFunc
	wg         sync.WaitGroup
}
type KafkaPipe struct {
	cfg            config.PipeConfig
	conn           *sql.DB
	saramaConsumer sarama.Consumer
	consumers      map[string]*topicConsumer
	lock           sync.RWMutex  
}
type kafkaProducer struct {
	topic    string
	producer sarama.SyncProducer
	batch    []*sarama.ProducerMessage
	batchPtr int
	log      log.Logger
}
type kafkaConsumer struct {
	baseConsumer
	pipe  *KafkaPipe
	topic string
	ch    chan *sarama.ConsumerMessage
	log   log.Logger
}
func init() {
	registerPlugin("kafka", initKafkaPipe)
}
func initKafkaPipe(cfg *config.PipeConfig, db *sql.DB) (Pipe, error) {
	return &KafkaPipe{conn: db, cfg: *cfg}, nil
}