Chore.mns no sleep 2

docker-compose.yml

@@ -51,3 +51,34 @@ services:
       - ETCD_ADVERTISE_CLIENT_URLS=http://mode-provider:2379
     ports:
       - "2379:2379"
+  zipkin:
+    image: openzipkin/zipkin:latest
+    restart: unless-stopped
+    ports:
+      - "9411:9411"
+    environment:
+      - STORAGE_TYPE=mem
+  redpanda:
+    image: redpandadata/redpanda:latest
+    container_name: redpanda
+    ports:
+      - "9092:9092"  # Kafka API
+    environment:
+      - REDPANDA_RPC_SERVER_LISTEN_ADDR=0.0.0.0
+      - REDPANDA_ADVERTISED_KAFKA_ADDR=localhost:9092
+    command:
+      - redpanda
+      - start
+      - --smp=1
+      - --memory=1G
+      - --reserve-memory=0M
+      - --overprovisioned
+      - --node-id=0
+      - --check=false
+      - --kafka-addr=0.0.0.0:9092
+      - --advertise-kafka-addr=localhost:9092
+      - --pandaproxy-addr=0.0.0.0:8082
+      - --advertise-pandaproxy-addr=localhost:8082
+      - --schema-registry-addr=0.0.0.0:8081
+      - --rpc-addr=0.0.0.0:33145
+      - --advertise-rpc-addr=localhost:33145

jobsdb/jobsdb.go

@@ -409,6 +409,9 @@ type JobT struct {
 	LastJobStatus JobStatusT      `json:"LastJobStatus"`
 	Parameters    json.RawMessage `json:"Parameters"`
 	WorkspaceId   string          `json:"WorkspaceId"`
+
+	ctx     context.Context
+	ctxOnce sync.Once
 }
 
 func (job *JobT) String() string {
@@ -428,6 +431,34 @@ func (job *JobT) sanitizeJSON() error {
 	return nil
 }
 
+type traceContainer struct {
+	TraceParent string `json:"traceparent"`
+}
+
+func (job *JobT) Ctx() context.Context {
+	job.ctxOnce.Do(func() {
+		if job.Parameters == nil {
+			job.ctx = context.Background()
+			return
+		}
+
+		var c traceContainer
+		err := jsonrs.Unmarshal(job.Parameters, &c)
+		if err != nil {
+			job.ctx = context.Background()
+			return
+		}
+
+		if c.TraceParent == "" {
+			job.ctx = context.Background()
+			return
+		}
+		ctx := stats.InjectTraceParentIntoContext(context.Background(), c.TraceParent)
+		job.ctx = ctx
+	})
+	return job.ctx
+}
+
 // The struct fields need to be exposed to JSON package
 type dataSetT struct {
 	JobTable       string `json:"job"`

processor/internal/transformer/destination_transformer/destination_transformer.go

@@ -54,6 +54,16 @@ func New(conf *config.Config, log logger.Logger, stat stats.Stats, opts ...Opt)
 	handle.config.maxRetryBackoffInterval = conf.GetReloadableDurationVar(30, time.Second, "Processor.DestinationTransformer.maxRetryBackoffInterval", "Processor.maxRetryBackoffInterval")
 	handle.config.batchSize = config.GetReloadableIntVar(100, 1, "Processor.DestinationTransformer.batchSize", "Processor.transformBatchSize")
 
+	handle.config.maxLoggedEvents = conf.GetReloadableIntVar(10000, 1, "Processor.DestinationTransformer.maxLoggedEvents")
+
+	handle.stats.comparisonTime = handle.stat.NewStat("embedded_destination_transform_comparison_time", stats.TimerType)
+	handle.stats.matchedEvents = handle.stat.NewStat("embedded_destination_transform_matched_events", stats.CountType)
+	handle.stats.mismatchedEvents = handle.stat.NewStat("embedded_destination_transform_mismatched_events", stats.CountType)
+
+	handle.loggedEvents = 0
+	handle.loggedEventsMu = sync.Mutex{}
+	handle.loggedFileName = generateLogFileName()
+
 	for _, opt := range opts {
 		opt(handle)
 	}
@@ -70,15 +80,27 @@ type Client struct {
 		maxRetryBackoffInterval config.ValueLoader[time.Duration]
 		timeoutDuration         time.Duration
 		batchSize               config.ValueLoader[int]
+
+		maxLoggedEvents config.ValueLoader[int]
 	}
 	guardConcurrency chan struct{}
 	conf             *config.Config
 	log              logger.Logger
 	stat             stats.Stats
 	client           transformerclient.Client
+
+	stats struct {
+		comparisonTime   stats.Timer
+		matchedEvents    stats.Counter
+		mismatchedEvents stats.Counter
+	}
+
+	loggedEventsMu sync.Mutex
+	loggedEvents   int64
+	loggedFileName string
 }
 
-func (d *Client) Transform(ctx context.Context, clientEvents []types.TransformerEvent) types.Response {
+func (d *Client) transform(ctx context.Context, clientEvents []types.TransformerEvent) types.Response {
 	batchSize := d.config.batchSize.Load()
 	if len(clientEvents) == 0 {
 		return types.Response{}

processor/internal/transformer/destination_transformer/embedded/kafka/kafka.go

@@ -0,0 +1,123 @@
+package kafka
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	backendconfig "github.com/rudderlabs/rudder-server/backend-config"
+	utils "github.com/rudderlabs/rudder-server/processor/internal/transformer/destination_transformer/embedded"
+	"github.com/rudderlabs/rudder-server/processor/types"
+	"github.com/rudderlabs/rudder-server/utils/misc"
+)
+
+var canonicalNames = []string{"KAFKA", "kafka", "Kafka"}
+
+func Transform(_ context.Context, events []types.TransformerEvent) types.Response {
+	response := types.Response{}
+	eventTypeToTopicMap := utils.GetTopicMap(events[0].Destination, "eventTypeToTopicMap", true)
+	eventToTopicMap := utils.GetTopicMap(events[0].Destination, "eventToTopicMap", false)
+
+	for _, event := range events {
+		event.Metadata.SourceDefinitionType = "" // TODO: Currently, it's getting ignored during JSON marshalling Remove this once we start using it.
+
+		if event.Destination.ID != events[0].Destination.ID {
+			panic("all events must have the same destination")
+		}
+
+		event.Message = utils.UpdateTimestampFieldForRETLEvent(event.Message)
+		var integrationsObj map[string]interface{}
+		for _, canonicalName := range canonicalNames {
+			if inObj, ok := misc.MapLookup(event.Message, "integrations", canonicalName).(map[string]interface{}); ok {
+				integrationsObj = inObj
+				break
+			}
+		}
+
+		var userId string
+		if id, ok := event.Message["userId"].(string); ok && id != "" {
+			userId = id
+		} else if id, ok := event.Message["anonymousId"].(string); ok {
+			userId = id
+		}
+
+		topic, err := getTopic(event, integrationsObj, eventTypeToTopicMap, eventToTopicMap)
+		if err != nil {
+			response.FailedEvents = append(response.FailedEvents, types.TransformerResponse{
+				Error:      err.Error(),
+				Metadata:   event.Metadata,
+				StatusCode: http.StatusInternalServerError,
+				StatTags:   utils.GetValidationErrorStatTags(event.Destination),
+			})
+			continue
+		}
+
+		outputEvent := map[string]interface{}{
+			"message": utils.GetMessageAsMap(event.Message),
+			"userId":  userId,
+			"topic":   topic,
+		}
+
+		if schemaId, ok := integrationsObj["schemaId"].(string); ok && schemaId != "" {
+			outputEvent["schemaId"] = schemaId
+		}
+
+		event.Metadata.RudderID = fmt.Sprintf("%s<<>>%s", event.Metadata.RudderID, topic)
+
+		response.Events = append(response.Events, types.TransformerResponse{
+			Output:     outputEvent,
+			Metadata:   event.Metadata,
+			StatusCode: http.StatusOK,
+		})
+	}
+
+	return response
+}
+
+func getTopic(event types.TransformerEvent, integrationsObj map[string]interface{}, eventTypeToTopicMap, eventToTopicMap map[string]string) (string, error) {
+	if topic, ok := integrationsObj["topic"].(string); ok && topic != "" {
+		return topic, nil
+	}
+
+	if topic, ok := filterConfigTopics(event.Message, event.Destination, eventTypeToTopicMap, eventToTopicMap); ok && topic != "" {
+		return topic, nil
+	}
+
+	if topic, ok := event.Destination.Config["topic"].(string); ok && topic != "" {
+		return topic, nil
+	}
+
+	return "", fmt.Errorf("Topic is required for Kafka destination")
+}
+
+func filterConfigTopics(message types.SingularEventT, destination backendconfig.DestinationT, eventTypeToTopicMap, eventToTopicMap map[string]string) (string, bool) {
+	if destination.Config["enableMultiTopic"] == true {
+		messageType, ok := message["type"].(string)
+		if !ok {
+			return "", false
+		}
+
+		switch messageType {
+		case "identify", "screen", "page", "group", "alias":
+			{
+				if topic, ok := eventTypeToTopicMap[messageType]; ok {
+					return topic, true
+				}
+				break
+			}
+		case "track":
+			{
+				eventName, ok := message["event"].(string)
+				if !ok || eventName == "" {
+					return "", false
+				}
+
+				if topic, ok := eventToTopicMap[eventName]; ok {
+					return topic, true
+				}
+				break
+			}
+		}
+	}
+	return "", false
+}