Updated Kafka message

cmd/utils/flaggroup.go

@@ -144,6 +144,8 @@ var FlagGroups = []FlagGroup{
 			ChainDataFetcherKafkaMaxMessageBytesFlag,
 			ChainDataFetcherKafkaSegmentSizeBytesFlag,
 			ChainDataFetcherKafkaRequiredAcksFlag,
+			ChainDataFetcherKafkaMessageVersionFlag,
+			ChainDataFetcherKafkaProducerIdFlag,
 		},
 	},
 	{

cmd/utils/flags.go

@@ -840,6 +840,16 @@ var (
 		Usage: "The level of acknowledgement reliability needed from Kafka broker (0: NoResponse, 1: WaitForLocal, -1: WaitForAll)",
 		Value: kafka.DefaultRequiredAcks,
 	}
+	ChainDataFetcherKafkaMessageVersionFlag = cli.StringFlag{
+		Name:  "chaindatafetcher.kafka.msg.version",
+		Usage: "The version of Kafka message",
+		Value: kafka.DefaultKafkaMessageVersion,
+	}
+	ChainDataFetcherKafkaProducerIdFlag = cli.StringFlag{
+		Name:  "chaindatafetcher.kafka.producer.id",
+		Usage: "The identifier of kafka message producer",
+		Value: kafka.GetDefaultProducerId(),
+	}
 	// DBSyncer
 	EnableDBSyncerFlag = cli.BoolFlag{
 		Name:  "dbsyncer",

cmd/utils/nodecmd/dumpconfigcmd.go

@@ -229,6 +229,8 @@ func makeKafkaConfig(ctx *cli.Context) *kafka.KafkaConfig {
 	kafkaConfig.Replicas = int16(ctx.GlobalInt64(utils.ChainDataFetcherKafkaReplicasFlag.Name))
 	kafkaConfig.SaramaConfig.Producer.MaxMessageBytes = ctx.GlobalInt(utils.ChainDataFetcherKafkaMaxMessageBytesFlag.Name)
 	kafkaConfig.SegmentSizeBytes = ctx.GlobalInt(utils.ChainDataFetcherKafkaSegmentSizeBytesFlag.Name)
+	kafkaConfig.MsgVersion = ctx.GlobalString(utils.ChainDataFetcherKafkaMessageVersionFlag.Name)
+	kafkaConfig.ProducerId = ctx.GlobalString(utils.ChainDataFetcherKafkaProducerIdFlag.Name)
 	requiredAcks := sarama.RequiredAcks(ctx.GlobalInt(utils.ChainDataFetcherKafkaRequiredAcksFlag.Name))
 	if requiredAcks != sarama.NoResponse && requiredAcks != sarama.WaitForLocal && requiredAcks != sarama.WaitForAll {
 		logger.Crit("not supported requiredAcks. it must be NoResponse(0), WaitForLocal(1), or WaitForAll(-1)", "given", requiredAcks)

cmd/utils/nodecmd/nodeflags.go

@@ -185,6 +185,8 @@ var KENFlags = []cli.Flag{
 	utils.ChainDataFetcherKafkaMaxMessageBytesFlag,
 	utils.ChainDataFetcherKafkaSegmentSizeBytesFlag,
 	utils.ChainDataFetcherKafkaRequiredAcksFlag,
+	utils.ChainDataFetcherKafkaMessageVersionFlag,
+	utils.ChainDataFetcherKafkaProducerIdFlag,
 	// DBSyncer
 	utils.EnableDBSyncerFlag,
 	utils.DBHostFlag,

datasync/chaindatafetcher/kafka/config.go

@@ -20,6 +20,8 @@ import (
 	"fmt"
 
 	"github.com/Shopify/sarama"
+	"github.com/klaytn/klaytn/common"
+	"github.com/klaytn/klaytn/common/hexutil"
 )
 
 const (
@@ -42,10 +44,14 @@ const (
 	DefaultRequiredAcks         = 1
 	DefaultSegmentSizeBytes     = 1000000 // 1 MB
 	DefaultMaxMessageNumber     = 100     // max number of messages in buffer
+	DefaultKafkaMessageVersion  = MsgVersion1_0
+	DefaultProducerIdPrefix     = "producer-"
 )
 
 type KafkaConfig struct {
 	SaramaConfig         *sarama.Config `json:"-"` // kafka client configurations.
+	MsgVersion           string         // MsgVersion is the version of Kafka message.
+	ProducerId           string         // ProducerId is for the identification of the message publisher.
 	Brokers              []string       // Brokers is a list of broker URLs.
 	TopicEnvironmentName string
 	TopicResourceName    string
@@ -74,9 +80,17 @@ func GetDefaultKafkaConfig() *KafkaConfig {
 		Replicas:             DefaultReplicas,
 		SegmentSizeBytes:     DefaultSegmentSizeBytes,
 		MaxMessageNumber:     DefaultMaxMessageNumber,
+		MsgVersion:           DefaultKafkaMessageVersion,
+		ProducerId:           GetDefaultProducerId(),
 	}
 }
 
+func GetDefaultProducerId() string {
+	rb := common.MakeRandomBytes(8)
+	randomString := hexutil.Encode(rb)
+	return DefaultProducerIdPrefix + randomString[2:]
+}
+
 func (c *KafkaConfig) GetTopicName(event string) string {
 	return fmt.Sprintf("%v.%v.%v.%v.%v.%v", c.TopicEnvironmentName, topicProjectName, topicServiceName, c.TopicResourceName, event, topicVersion)
 }

datasync/chaindatafetcher/kafka/consumer.go

@@ -37,6 +37,7 @@ var (
 	nilConsumerMessageErrorMsg = "the given message should not be nil"
 	wrongHeaderNumberErrorMsg  = "the number of header is not expected"
 	wrongHeaderKeyErrorMsg     = "the header key is not expected"
+	wrongMsgVersionErrorMsg    = "the message version is not supported"
 	missingSegmentErrorMsg     = "there is a missing segment"
 	noHandlerErrorMsg          = "the handler does not exist for the given topic"
 	emptySegmentErrorMsg       = "there is no segment in the segment slice"
@@ -48,15 +49,17 @@ type TopicHandler func(message *sarama.ConsumerMessage) error
 
 // Segment represents a message segment with the parsed headers.
 type Segment struct {
-	orig  *sarama.ConsumerMessage
-	key   string
-	total uint64
-	index uint64
-	value []byte
+	orig       *sarama.ConsumerMessage
+	key        string
+	total      uint64
+	index      uint64
+	value      []byte
+	version    string
+	producerId string
 }
 
 func (s *Segment) String() string {
-	return fmt.Sprintf("key: %v, total: %v, index: %v, value %v", s.key, s.total, s.index, string(s.value))
+	return fmt.Sprintf("key: %v, total: %v, index: %v, value: %v, version: %v, producerId: %v", s.key, s.total, s.index, string(s.value), s.version, s.producerId)
 }
 
 // newSegment creates a new segment structure after parsing the headers.
@@ -65,8 +68,30 @@ func newSegment(msg *sarama.ConsumerMessage) (*Segment, error) {
 		return nil, errors.New(nilConsumerMessageErrorMsg)
 	}
 
-	if len(msg.Headers) != MsgHeaderLength {
-		return nil, fmt.Errorf("%v [header length: %v]", wrongHeaderNumberErrorMsg, len(msg.Headers))
+	headerLen := len(msg.Headers)
+	if headerLen != MsgHeaderLength && headerLen != LegacyMsgHeaderLength {
+		return nil, fmt.Errorf("%v [header length: %v]", wrongHeaderNumberErrorMsg, headerLen)
+	}
+
+	version := ""
+	producerId := ""
+
+	if len(msg.Headers) == MsgHeaderLength {
+		keyVersion := string(msg.Headers[MsgHeaderVersion].Key)
+		if keyVersion != KeyVersion {
+			return nil, fmt.Errorf("%v [expected: %v, actual: %v]", wrongHeaderKeyErrorMsg, KeyVersion, keyVersion)
+		}
+		version = string(msg.Headers[MsgHeaderVersion].Value)
+		switch version {
+		case MsgVersion1_0:
+			keyProducerId := string(msg.Headers[MsgHeaderProducerId].Key)
+			if keyProducerId != KeyProducerId {
+				return nil, fmt.Errorf("%v [expected: %v, actual: %v]", wrongHeaderKeyErrorMsg, KeyProducerId, keyProducerId)
+			}
+			producerId = string(msg.Headers[MsgHeaderProducerId].Value)
+		default:
+			return nil, fmt.Errorf("%v [available: %v]", wrongMsgVersionErrorMsg, MsgVersion1_0)
+		}
 	}
 
 	// check the existence of KeyTotalSegments header
@@ -84,7 +109,15 @@ func newSegment(msg *sarama.ConsumerMessage) (*Segment, error) {
 	key := string(msg.Key)
 	totalSegments := binary.BigEndian.Uint64(msg.Headers[MsgHeaderTotalSegments].Value)
 	segmentIdx := binary.BigEndian.Uint64(msg.Headers[MsgHeaderSegmentIdx].Value)
-	return &Segment{orig: msg, key: key, total: totalSegments, index: segmentIdx, value: msg.Value}, nil
+	return &Segment{
+		orig:       msg,
+		key:        key,
+		total:      totalSegments,
+		index:      segmentIdx,
+		value:      msg.Value,
+		version:    version,
+		producerId: producerId,
+	}, nil
 }
 
 // Consumer is a reference structure to subscribe block or trace group produced by EN.
@@ -174,7 +207,7 @@ func (c *Consumer) Cleanup(s sarama.ConsumerGroupSession) error {
 func insertSegment(newSegment *Segment, buffer [][]*Segment) ([][]*Segment, error) {
 	for idx, bufferedSegments := range buffer {
 		numBuffered := len(bufferedSegments)
-		if numBuffered > 0 && bufferedSegments[0].key == newSegment.key {
+		if numBuffered > 0 && bufferedSegments[0].key == newSegment.key && bufferedSegments[0].producerId == newSegment.producerId {
 			// there is a missing segment which should not exist.
 			if newSegment.index > uint64(numBuffered) {
 				logger.Error("there may be a missing segment", "numBuffered", numBuffered, "newSegment", newSegment)