update

cdc/sink/mq.go

@@ -143,18 +143,18 @@ func (k *mqSink) calPartition(row *model.RowChangedEvent) int32 {
 		log.Fatal("calculate hash of message key failed, please report a bug", zap.Error(err))
 	}
 
-	if len(row.IndieMarkCol) > 0 {
-		// distribute partition by rowid or unique column value
-		value := row.Columns[row.IndieMarkCol].Value
-		b, err := json.Marshal(value)
-		if err != nil {
-			log.Fatal("calculate hash of message key failed, please report a bug", zap.Error(err))
-		}
-		_, err = hash.Write(b)
-		if err != nil {
-			log.Fatal("calculate hash of message key failed, please report a bug", zap.Error(err))
-		}
+	//if len(row.IndieMarkCol) > 0 {
+	// distribute partition by rowid or unique column value
+	value := row.Columns[row.IndieMarkCol].Value
+	b, err := json.Marshal(value)
+	if err != nil {
+		log.Fatal("calculate hash of message key failed, please report a bug", zap.Error(err))
 	}
+	_, err = hash.Write(b)
+	if err != nil {
+		log.Fatal("calculate hash of message key failed, please report a bug", zap.Error(err))
+	}
+
 	return int32(hash.Sum32() % uint32(k.partitionNum))
 }
 

cdc/sink/mqProducer/kafka.go

@@ -73,18 +73,18 @@ func (k *kafkaSaramaProducer) SendMessage(ctx context.Context, key []byte, value
 			callback(err)
 		}
 	}
-	cb(nil)
-	//select {
-	//case <-ctx.Done():
-	//	return 0, errors.Trace(ctx.Err())
-	//case k.asyncClient.Input() <- &sarama.ProducerMessage{
-	//	Topic:     k.topic,
-	//	Key:       sarama.ByteEncoder(key),
-	//	Value:     sarama.ByteEncoder(value),
-	//	Partition: partition,
-	//	Metadata:  cb,
-	//}:
-	//}
+	//cb(nil)
+	select {
+	case <-ctx.Done():
+		return 0, errors.Trace(ctx.Err())
+	case k.asyncClient.Input() <- &sarama.ProducerMessage{
+		Topic:     k.topic,
+		Key:       sarama.ByteEncoder(key),
+		Value:     sarama.ByteEncoder(value),
+		Partition: partition,
+		Metadata:  cb,
+	}:
+	}
 	return index, nil
 }