Repository: knative-sandbox/eventing-kafka
Pull request #1001 (2021-11-24 18:36:55)

[KafkaChannels] Pass-through "additional headers"
The common `MessageReceiver` implementation forwards "additional headers" to the registered handler.  These are **non-ce** headers that are deemed important to pass-through (x-b3, etc).  Currently the KafkaChannel implementations do **not** forward these headers.  Not forwarding such headers can, as an example, limit the ability to perform complete e2e tracing in Istio.

## Proposed Changes

- 🎁 Enhance receiver to append "additional headers" into Sarama `ProducerMessage.Headers`
- 🎁 Enhance dispatcher to extract "additional headers" from Sarama `ConsumerMessage.Headers`
- 🎁 Enhance dispatcher to include "additional headers" in outbound requests.

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
KafkaChannels will now pass-through the "x-request-id", "knative-*" and "x-b3-*" headers for all events.
```


knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/1001#" title="Author self-approved">travis-minke-sap</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/main/OWNERS)~~ [travis-minke-sap]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/1001/pull-knative-sandbox-eventing-kafka-go-coverage/1463577388905402368/artifacts/line-cov.html#file1) | 66.1% | 65.5% | -0.6
[pkg/channel/distributed/dispatcher/dispatcher/handler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/1001/pull-knative-sandbox-eventing-kafka-go-coverage/1463577388905402368/artifacts/line-cov.html#file2) | 92.3% | 92.5% | 0.2
[pkg/channel/distributed/receiver/producer/producer.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/1001/pull-knative-sandbox-eventing-kafka-go-coverage/1463577388905402368/artifacts/line-cov.html#file3) | 78.1% | 78.4% | 0.3

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1001?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1001](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1001?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (90e8501) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/b50f5d4e986f3f11bd79892b1cf378f16a9daee5?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (b50f5d4) will **increase** coverage by `0.04%`.
> The diff coverage is `85.00%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1001/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1001?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@            Coverage Diff             @@
##             main    #1001      +/-   ##
==========================================
+ Coverage   74.69%   74.74%   +0.04%     
==========================================
  Files         121      121              
  Lines        5675     5693      +18     
==========================================
+ Hits         4239     4255      +16     
- Misses       1213     1215       +2     
  Partials      223      223              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1001?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...onsolidated/dispatcher/consumer\_message\_handler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1001/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvY29uc3VtZXJfbWVzc2FnZV9oYW5kbGVyLmdv) | `0.00% <0.00%> (ø)` | |
| [pkg/channel/consolidated/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1001/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvZGlzcGF0Y2hlci5nbw==) | `63.69% <0.00%> (-0.44%)` | :arrow_down: |
| [...annel/distributed/dispatcher/dispatcher/handler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1001/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2hhbmRsZXIuZ28=) | `89.09% <100.00%> (+0.20%)` | :arrow_up: |
| [.../channel/distributed/receiver/producer/producer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1001/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvcmVjZWl2ZXIvcHJvZHVjZXIvcHJvZHVjZXIuZ28=) | `69.66% <100.00%> (+0.34%)` | :arrow_up: |
| [pkg/common/tracing/tracing.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1001/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi90cmFjaW5nL3RyYWNpbmcuZ28=) | `100.00% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1001?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1001?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [b50f5d4...90e8501](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1001?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/1001/pull-knative-sandbox-eventing-kafka-go-coverage/1465385246215114752/artifacts/line-cov.html#file1) | 66.1% | 65.5% | -0.6
[pkg/channel/distributed/dispatcher/dispatcher/handler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/1001/pull-knative-sandbox-eventing-kafka-go-coverage/1465385246215114752/artifacts/line-cov.html#file2) | 92.3% | 92.5% | 0.2
[pkg/channel/distributed/receiver/producer/producer.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/1001/pull-knative-sandbox-eventing-kafka-go-coverage/1465385246215114752/artifacts/line-cov.html#file3) | 78.1% | 78.4% | 0.3


Patch:
diff --git a/cmd/channel/distributed/receiver/main.go b/cmd/channel/distributed/receiver/main.go
index 65f7063ee..28d0f9947 100644
--- a/cmd/channel/distributed/receiver/main.go
+++ b/cmd/channel/distributed/receiver/main.go
@@ -182,7 +182,7 @@ func flush(logger *zap.Logger) {
 }
 
 // CloudEvent Message Handler - Converts To KafkaMessage And Produces To Channel's Kafka Topic
-func handleMessage(ctx context.Context, channelReference eventingchannel.ChannelReference, message binding.Message, transformers []binding.Transformer, _ http.Header) error {
+func handleMessage(ctx context.Context, channelReference eventingchannel.ChannelReference, message binding.Message, transformers []binding.Transformer, httpHeader http.Header) error {
 
 	// Get The Logger From The Context
 	logger := logging.FromContext(ctx).Desugar()
@@ -205,7 +205,7 @@ func handleMessage(ctx context.Context, channelReference eventingchannel.Channel
 	}
 
 	// Produce The CloudEvent Binding Message (Send To The Appropriate Kafka Topic)
-	err = kafkaProducer.ProduceKafkaMessage(ctx, channelReference, message, transformers...)
+	err = kafkaProducer.ProduceKafkaMessage(ctx, channelReference, message, httpHeader, transformers...)
 	if err != nil {
 		logger.Error("Failed To Produce Kafka Message", zap.Error(err))
 		return err
diff --git a/go.mod b/go.mod
index c160a3b8b..23eac2ab8 100644
--- a/go.mod
+++ b/go.mod
@@ -33,9 +33,9 @@ require (
 	k8s.io/client-go v0.21.4
 	k8s.io/utils v0.0.0-20201110183641-67b214c5f920
 	knative.dev/control-protocol v0.0.0-20211115173555-0910714d1183
-	knative.dev/eventing v0.27.1-0.20211123205351-820db20be4b2
+	knative.dev/eventing v0.27.1-0.20211126120551-7fc053b79089
 	knative.dev/hack v0.0.0-20211122162614-813559cefdda
-	knative.dev/pkg v0.0.0-20211123135150-787aec59e70a
+	knative.dev/pkg v0.0.0-20211129195804-438776b3c87c
 	knative.dev/reconciler-test v0.0.0-20211112132636-ae9e2e21972f
 	sigs.k8s.io/yaml v1.3.0
 )
diff --git a/go.sum b/go.sum
index 093b59e3b..5c6dd9ed5 100644
--- a/go.sum
+++ b/go.sum
@@ -1328,8 +1328,8 @@ k8s.io/utils v0.0.0-20201110183641-67b214c5f920 h1:CbnUZsM497iRC5QMVkHwyl8s2tB3g
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20211115173555-0910714d1183 h1:0Rvhwk9dfIqNGBkZWmxFNvcwYVrnphHZRXolSc5UDsw=
 knative.dev/control-protocol v0.0.0-20211115173555-0910714d1183/go.mod h1:h/Ac2W8Db0pgTo4baEfzRhn+MPpxJnkSQTBTZjthwXY=
-knative.dev/eventing v0.27.1-0.20211123205351-820db20be4b2 h1:QwyD6phsrqx5hDplMkFZ6NJPLMSnxN3YjAw+Qnnrbpg=
-knative.dev/eventing v0.27.1-0.20211123205351-820db20be4b2/go.mod h1:fnBpTKtPOxC7WOaQY6WUiUAVBXTb44njuhTi7LXHGVc=
+knative.dev/eventing v0.27.1-0.20211126120551-7fc053b79089 h1:CxTQ4uk9BpwWzXARM/20uENtwGwxOyS11ankbuz9GDE=
+knative.dev/eventing v0.27.1-0.20211126120551-7fc053b79089/go.mod h1:gGQdsi8sNQKIfCFJ58CngPFBrnO5Oqn8pNjZNFMDg88=
 knative.dev/hack v0.0.0-20211101195839-11d193bf617b/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211112192837-128cf0150a69/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211122162614-813559cefdda h1:WBfGcmu5pN+RCxDotKVYtWlJVI2V4Hvah+XD9mcSR/o=
@@ -1337,9 +1337,9 @@ knative.dev/hack v0.0.0-20211122162614-813559cefdda/go.mod h1:PHt8x8yX5Z9pPquBEf
 knative.dev/hack/schema v0.0.0-20211122162614-813559cefdda/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc/go.mod h1:SkfDk9bWIiNZD7XtILGkG7AKVyF/M6M0bGxLgl0SYL8=
 knative.dev/pkg v0.0.0-20211115071955-517ef0292b53/go.mod h1:VqUp1KWJqpTDNoiSI/heaX3uMdubImslJE2tBkP+Bbw=
-knative.dev/pkg v0.0.0-20211120133512-d016976f2567/go.mod h1:VqUp1KWJqpTDNoiSI/heaX3uMdubImslJE2tBkP+Bbw=
-knative.dev/pkg v0.0.0-20211123135150-787aec59e70a h1:mgE9GdkjfPJc7KpgBox52dSy1GsQmsFTWXy+1HjEn9I=
-knative.dev/pkg v0.0.0-20211123135150-787aec59e70a/go.mod h1:fZUlVceKtVNyFU6LokWGk2a9QFQXeSTtnbKOjjkQ690=
+knative.dev/pkg v0.0.0-20211125172117-608fc877e946/go.mod h1:fZUlVceKtVNyFU6LokWGk2a9QFQXeSTtnbKOjjkQ690=
+knative.dev/pkg v0.0.0-20211129195804-438776b3c87c h1:TcVqa/i/E/KcusxjroLEaMB2bDV42lYzN2Z+4soudE8=
+knative.dev/pkg v0.0.0-20211129195804-438776b3c87c/go.mod h1:AKPae1Cmj+k0GWXWnF2tKY7q5qPa1mTD7oCP4OeMvEM=
 knative.dev/reconciler-test v0.0.0-20211112132636-ae9e2e21972f h1:bSdsp5av09LqSF4iXo8LCCDEENJ+JEbzzdgHBoJdf2w=
 knative.dev/reconciler-test v0.0.0-20211112132636-ae9e2e21972f/go.mod h1:gTsbLk496j/M9xqMpx/liyCQ0X3bwDpRtcs2Zzws364=
 pgregory.net/rapid v0.3.3 h1:jCjBsY4ln4Atz78QoBWxUEvAHaFyNDQg9+WU62aCn1U=
diff --git a/pkg/channel/consolidated/dispatcher/consumer_message_handler.go b/pkg/channel/consolidated/dispatcher/consumer_message_handler.go
index 641db4fd2..599a30193 100644
--- a/pkg/channel/consolidated/dispatcher/consumer_message_handler.go
+++ b/pkg/channel/consolidated/dispatcher/consumer_message_handler.go
@@ -24,11 +24,12 @@ import (
 	protocolkafka "github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2"
 	"github.com/cloudevents/sdk-go/v2/binding"
 	"go.uber.org/zap"
-	"knative.dev/eventing-kafka/pkg/common/consumer"
-	"knative.dev/eventing-kafka/pkg/common/tracing"
 	eventingchannels "knative.dev/eventing/pkg/channel"
-	fanout "knative.dev/eventing/pkg/channel/fanout"
+	"knative.dev/eventing/pkg/channel/fanout"
 	"knative.dev/eventing/pkg/kncloudevents"
+
+	"knative.dev/eventing-kafka/pkg/common/consumer"
+	"knative.dev/eventing-kafka/pkg/common/tracing"
 )
 
 type consumerMessageHandler struct {
@@ -70,6 +71,11 @@ func (c consumerMessageHandler) Handle(ctx context.Context, consumerMessage *sar
 		zap.String("subscription", c.sub.String()),
 	)
 
+	// Convert ConsumerMessage.Headers Into HTTP Header Struct For Dispatching (Passing-Through of "Additional Headers")
+	// Using Sarama RecordHeaders instead of CloudEvent Message.Headers to support multi-value HTTP Headers without
+	// serialization.  Also, filtering CloudEvent "ce" headers which are already taken from the Message.
+	httpHeader := tracing.ConvertRecordHeadersToHttpHeader(tracing.FilterCeRecordHeaders(consumerMessage.Headers))
+
 	ctx, span := tracing.StartTraceFromMessage(c.logger, ctx, message, consumerMessage.Topic)
 	defer span.End()
 
@@ -78,7 +84,7 @@ func (c consumerMessageHandler) Handle(ctx context.Context, consumerMessage *sar
 	dispatchExecutionInfo, err := c.dispatcher.DispatchMessageWithRetries(
 		ctx,
 		message,
-		nil,
+		httpHeader,
 		c.sub.Subscriber,
 		c.sub.Reply,
 		c.sub.DeadLetter,
diff --git a/pkg/channel/consolidated/dispatcher/dispatcher.go b/pkg/channel/consolidated/dispatcher/dispatcher.go
index bd1e7cf2a..092f33da4 100644
--- a/pkg/channel/consolidated/dispatcher/dispatcher.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher.go
@@ -104,7 +104,7 @@ func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs, enqueue func(
 	reporter := eventingchannels.NewStatsReporter(containerName, kmeta.ChildName(podName, uuid.New().String()))
 	dispatcher.reporter = reporter
 	receiverFunc, err := eventingchannels.NewMessageReceiver(
-		func(ctx context.Context, channel eventingchannels.ChannelReference, message binding.Message, transformers []binding.Transformer, _ nethttp.Header) error {
+		func(ctx context.Context, channel eventingchannels.ChannelReference, message binding.Message, transformers []binding.Transformer, httpHeader nethttp.Header) error {
 			kafkaProducerMessage := sarama.ProducerMessage{
 				Topic: dispatcher.topicFunc(utils.KafkaChannelSeparator, channel.Namespace, channel.Name),
 			}
@@ -116,6 +116,7 @@ func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs, enqueue func(
 			}
 
 			kafkaProducerMessage.Headers = append(kafkaProducerMessage.Headers, tracing.SerializeTrace(trace.FromContext(ctx).SpanContext())...)
+			kafkaProducerMessage.Headers = append(kafkaProducerMessage.Headers, tracing.ConvertHttpHeaderToRecordHeaders(httpHeader)...)
 
 			partition, offset, err := dispatcher.kafkaSyncProducer.SendMessage(&kafkaProducerMessage)
 
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/handler.go b/pkg/channel/distributed/dispatcher/dispatcher/handler.go
index edd33bd2d..8e58e0dcd 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/handler.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/handler.go
@@ -122,6 +122,11 @@ func (h *Handler) Handle(ctx context.Context, consumerMessage *sarama.ConsumerMe
 			zap.Int64("Offset", consumerMessage.Offset))
 	}
 
+	// Convert ConsumerMessage.Headers Into HTTP Header Struct For Dispatching (Passing-Through of "Additional Headers")
+	// Using Sarama RecordHeaders instead of CloudEvent Message.Headers to support multi-value HTTP Headers without
+	// serialization.  Also, filtering CloudEvent "ce" headers which are already taken from the Message.
+	httpHeader := tracing.ConvertRecordHeadersToHttpHeader(tracing.FilterCeRecordHeaders(consumerMessage.Headers))
+
 	// Convert The Sarama ConsumerMessage Into A CloudEvents Message
 	message := kafkasaramaprotocol.NewMessageFromConsumerMessage(consumerMessage)
 	if message.ReadEncoding() == binding.EncodingUnknown {
@@ -134,7 +139,7 @@ func (h *Handler) Handle(ctx context.Context, consumerMessage *sarama.ConsumerMe
 	defer span.End()
 
 	// Dispatch The Message With Configured Retries, DLQ, etc
-	info, err := h.MessageDispatcher.DispatchMessageWithRetries(ctx, message, nil, h.destinationURL, h.replyURL, h.deadLetterURL, &h.retryConfig)
+	info, err := h.MessageDispatcher.DispatchMessageWithRetries(ctx, message, httpHeader, h.destinationURL, h.replyURL, h.deadLetterURL, &h.retryConfig)
 	h.Logger.Debug("Received Response", zap.Any("ExecutionInfo", executionInfoWrapper{info}))
 
 	//
diff --git a/pkg/channel/distributed/receiver/producer/producer.go b/pkg/channel/distributed/receiver/producer/producer.go
index e454d3c1d..87a296bde 100644
--- a/pkg/channel/distributed/receiver/producer/producer.go
+++ b/pkg/channel/distributed/receiver/producer/producer.go
@@ -19,6 +19,7 @@ package producer
 import (
 	"context"
 	"errors"
+	"net/http"
 	"time"
 
 	"github.com/Shopify/sarama"
@@ -28,6 +29,8 @@ import (
 	"go.opencensus.io/trace"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	eventingChannel "knative.dev/eventing/pkg/channel"
+
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/producer"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/receiver/constants"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/receiver/health"
@@ -37,7 +40,6 @@ import (
 	kafkasarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
 	"knative.dev/eventing-kafka/pkg/common/metrics"
 	"knative.dev/eventing-kafka/pkg/common/tracing"
-	eventingChannel "knative.dev/eventing/pkg/channel"
 )
 
 // Producer Struct
@@ -53,7 +55,7 @@ type Producer struct {
 	brokers            []string
 }
 
-// Initialize The Producer
+// NewProducer returns a new Producer instance with specified configuration.
 func NewProducer(logger *zap.Logger,
 	config *sarama.Config,
 	brokers []string,
@@ -94,8 +96,8 @@ func NewProducer(logger *zap.Logger,
 	return newProducer, nil
 }
 
-// Produce A KafkaMessage From The Specified CloudEvent To The Specified Topic And Wait For The Delivery Report
-func (p *Producer) ProduceKafkaMessage(ctx context.Context, channelReference eventingChannel.ChannelReference, message binding.Message, transformers ...binding.Transformer) error {
+// ProduceKafkaMessage creates and sends a Sarama ProducerMessage to the specified Topic and waits for the delivery confirmation.
+func (p *Producer) ProduceKafkaMessage(ctx context.Context, channelReference eventingChannel.ChannelReference, message binding.Message, httpHeader http.Header, transformers ...binding.Transformer) error {
 
 	// Validate The Kafka Producer (Must Be Pre-Initialized)
 	if p.kafkaProducer == nil {
@@ -120,6 +122,9 @@ func (p *Producer) ProduceKafkaMessage(ctx context.Context, channelReference eve
 	// Add The "traceparent" And "tracestate" Headers To The Message (Helps Tie Related Messages Together In Traces)
 	producerMessage.Headers = append(producerMessage.Headers, tracing.SerializeTrace(trace.FromContext(ctx).SpanContext())...)
 
+	// Add any additional headers ("x-b3", etc)
+	producerMessage.Headers = append(producerMessage.Headers, tracing.ConvertHttpHeaderToRecordHeaders(httpHeader)...)
+
 	// Produce The Kafka Message To The Kafka Topic
 	if logger.Core().Enabled(zap.DebugLevel) {
 		// Checked Logging Level First To Avoid Calling StringifyHeaders and Encode Functions In Production
@@ -139,7 +144,7 @@ func (p *Producer) ProduceKafkaMessage(ctx context.Context, channelReference eve
 	}
 }
 
-// Async Process For Observing Kafka Metrics
+// ObserveMetrics is an async process for observing Kafka metrics.
 func (p *Producer) ObserveMetrics(interval time.Duration) {
 
 	// Fork A New Process To Run Async Metrics Collection
@@ -221,7 +226,7 @@ func (p *Producer) SecretChanged(ctx context.Context, secret *corev1.Secret) *Pr
 	return reconfiguredKafkaProducer
 }
 
-// Close The Producer (Stop Processing)
+// Close gracefully stops all Producer processing.
 func (p *Producer) Close() {
 
 	// Mark The Producer As No Longer Ready
diff --git a/pkg/common/tracing/tracing.go b/pkg/common/tracing/tracing.go
index de5023023..4814b9609 100644
--- a/pkg/common/tracing/tracing.go
+++ b/pkg/common/tracing/tracing.go
@@ -18,6 +18,8 @@ package tracing
 
 import (
 	"context"
+	"net/http"
+	"strings"
 
 	"github.com/Shopify/sarama"
 	protocolkafka "github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2"
@@ -86,3 +88,47 @@ func ParseSpanContext(headers map[string][]byte) (sc trace.SpanContext, ok bool)
 
 	return format.SpanContextFromHeaders(traceParent, traceState)
 }
+
+// ConvertHttpHeaderToRecordHeaders converts the specified HTTP Header to Sarama RecordHeaders
+// It returns an array of RecordHeaders as is used in the Sarama ProducerMessage.
+// Multi-value HTTP Headers are decomposed into separate RecordHeader instances rather
+// than serializing as JSON or similar.
+func ConvertHttpHeaderToRecordHeaders(httpHeader http.Header) []sarama.RecordHeader {
+	recordHeaders := make([]sarama.RecordHeader, 0)
+	for headerKey, headerValues := range httpHeader {
+		for _, headerValue := range headerValues {
+			recordHeader := sarama.RecordHeader{
+				Key:   []byte(headerKey),
+				Value: []byte(headerValue),
+			}
+			recordHeaders = append(recordHeaders, recordHeader)
+		}
+	}
+	return recordHeaders
+}
+
+// ConvertRecordHeadersToHttpHeader converts the specified Sarama RecordHeaders to an HTTP Header.
+// It expects an array of RecordHeader pointers as is used in the Sarama ConsumerMessage.
+// Multi-value HTTP Headers are re-composed form RecordHeader instances sharing the same
+// Key rather than de-serializing JSON or similar.
+func ConvertRecordHeadersToHttpHeader(recordHeaders []*sarama.RecordHeader) http.Header {
+	httpHeader := make(http.Header)
+	for _, recordHeader := range recordHeaders {
+		if recordHeader != nil {
+			httpHeader.Add(string(recordHeader.Key), string(recordHeader.Value))
+		}
+	}
+	return httpHeader
+}
+
+// FilterCeRecordHeaders returns a new array of RecordHeader pointers including only
+// instances with non "ce_" Keys.
+func FilterCeRecordHeaders(recordHeaders []*sarama.RecordHeader) []*sarama.RecordHeader {
+	filteredRecordHeaders := make([]*sarama.RecordHeader, 0)
+	for _, recordHeader := range recordHeaders {
+		if recordHeader != nil && !strings.HasPrefix(string(recordHeader.Key), "ce_") {
+			filteredRecordHeaders = append(filteredRecordHeaders, recordHeader)
+		}
+	}
+	return filteredRecordHeaders
+}
diff --git a/vendor/knative.dev/pkg/configmap/parse.go b/vendor/knative.dev/pkg/configmap/parse.go
index faeacab14..e93f9af2a 100644
--- a/vendor/knative.dev/pkg/configmap/parse.go
+++ b/vendor/knative.dev/pkg/configmap/parse.go
@@ -53,6 +53,20 @@ func AsBool(key string, target *bool) ParseFunc {
 	}
 }
 
+// AsInt16 parses the value at key as an int16 into the target, if it exists.
+func AsInt16(key string, target *int16) ParseFunc {
+	return func(data map[string]string) error {
+		if raw, ok := data[key]; ok {
+			val, err := strconv.ParseInt(raw, 10, 16)
+			if err != nil {
+				return fmt.Errorf("failed to parse %q: %w", key, err)
+			}
+			*target = int16(val)
+		}
+		return nil
+	}
+}
+
 // AsInt32 parses the value at key as an int32 into the target, if it exists.
 func AsInt32(key string, target *int32) ParseFunc {
 	return func(data map[string]string) error {
@@ -95,6 +109,20 @@ func AsInt(key string, target *int) ParseFunc {
 	}
 }
 
+// AsUint16 parses the value at key as an uint16 into the target, if it exists.
+func AsUint16(key string, target *uint16) ParseFunc {
+	return func(data map[string]string) error {
+		if raw, ok := data[key]; ok {
+			val, err := strconv.ParseUint(raw, 10, 16)
+			if err != nil {
+				return fmt.Errorf("failed to parse %q: %w", key, err)
+			}
+			*target = uint16(val)
+		}
+		return nil
+	}
+}
+
 // AsUint32 parses the value at key as an uint32 into the target, if it exists.
 func AsUint32(key string, target *uint32) ParseFunc {
 	return func(data map[string]string) error {
diff --git a/vendor/modules.txt b/vendor/modules.txt
index b0db54b3f..509116cd9 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1187,7 +1187,7 @@ knative.dev/control-protocol/pkg/message
 knative.dev/control-protocol/pkg/network
 knative.dev/control-protocol/pkg/reconciler
 knative.dev/control-protocol/pkg/service
-# knative.dev/eventing v0.27.1-0.20211123205351-820db20be4b2
+# knative.dev/eventing v0.27.1-0.20211126120551-7fc053b79089
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/test
@@ -1329,7 +1329,7 @@ knative.dev/eventing/test/upgrade/prober/wathola/sender
 ## explicit
 knative.dev/hack
 knative.dev/hack/shell
-# knative.dev/pkg v0.0.0-20211123135150-787aec59e70a
+# knative.dev/pkg v0.0.0-20211129195804-438776b3c87c
 ## explicit
 knative.dev/pkg/apiextensions/storageversion
 knative.dev/pkg/apiextensions/storageversion/cmd/migrate


Test patch:
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/handler_test.go b/pkg/channel/distributed/dispatcher/dispatcher/handler_test.go
index 70767aad3..26187975c 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/handler_test.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/handler_test.go
@@ -16,6 +16,7 @@ package dispatcher
 import (
 	"context"
 	"fmt"
+	"net/http"
 	"net/url"
 	"strings"
 	"testing"
@@ -55,6 +56,7 @@ const (
 	testMsgEventTypeVersion  = "TestMsgEventTypeVersion"
 	testMsgKnativeHistory    = "TestKnativeHistory"
 	testMsgJsonContentString = "{\"content\": \"Test Message 1\"}"
+	testB3TraceId            = "TestB3TraceId"
 )
 
 var (
@@ -175,6 +177,12 @@ func TestGetConsumerGroup(t *testing.T) {
 // Test One Permutation Of The Handler's Handle() Functionality
 func performHandleTest(t *testing.T, testCase HandleTestCase) {
 
+	// Create The Expected Headers
+	headers := http.Header{
+		"Content-Type": []string{testMsgContentType},
+		"X-B3-Traceid": []string{testB3TraceId},
+	}
+
 	// Initialize Destination As Specified
 	var destinationUrl *url.URL
 	if testCase.destinationUri != nil {
@@ -206,7 +214,7 @@ func performHandleTest(t *testing.T, testCase HandleTestCase) {
 	}
 
 	// Create Mocks For Testing
-	mockMessageDispatcher := dispatchertesting.NewMockMessageDispatcher(t, nil, destinationUrl, replyUrl, deadLetterUrl, &retryConfig, testCase.dispatchErr)
+	mockMessageDispatcher := dispatchertesting.NewMockMessageDispatcher(t, headers, destinationUrl, replyUrl, deadLetterUrl, &retryConfig, testCase.dispatchErr)
 
 	// Mock The newMessageDispatcherWrapper Function (And Restore Post-Test)
 	newMessageDispatcherWrapperPlaceholder := newMessageDispatcherWrapper
@@ -336,6 +344,10 @@ func createConsumerMessage(t *testing.T) *sarama.ConsumerMessage {
 				Key:   []byte("ce_knativehistory"),
 				Value: []byte(testMsgKnativeHistory),
 			},
+			{
+				Key:   []byte("x-b3-traceid"),
+				Value: []byte(testB3TraceId),
+			},
 		},
 		Timestamp:      time.Now(),
 		BlockTimestamp: time.Time{},
diff --git a/pkg/channel/distributed/receiver/producer/producer_test.go b/pkg/channel/distributed/receiver/producer/producer_test.go
index 666ec48a5..15918359f 100644
--- a/pkg/channel/distributed/receiver/producer/producer_test.go
+++ b/pkg/channel/distributed/receiver/producer/producer_test.go
@@ -24,6 +24,9 @@ import (
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
+	"knative.dev/pkg/logging"
+	logtesting "knative.dev/pkg/logging/testing"
+
 	producertesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/producer/testing"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/receiver/constants"
 	channelhealth "knative.dev/eventing-kafka/pkg/channel/distributed/receiver/health"
@@ -33,8 +36,6 @@ import (
 	configtesting "knative.dev/eventing-kafka/pkg/common/config/testing"
 	"knative.dev/eventing-kafka/pkg/common/metrics"
 	commontesting "knative.dev/eventing-kafka/pkg/common/testing"
-	"knative.dev/pkg/logging"
-	logtesting "knative.dev/pkg/logging/testing"
 )
 
 // Test The NewProducer Constructor
@@ -66,6 +67,11 @@ func TestProduceKafkaMessage(t *testing.T) {
 	config := sarama.NewConfig()
 	channelReference := receivertesting.CreateChannelReference(receivertesting.ChannelName, receivertesting.ChannelNamespace)
 	bindingMessage := receivertesting.CreateBindingMessage(cloudevents.VersionV1)
+	httpHeader := map[string][]string{
+		"x-request-id": {"TestRequestId"},
+		"x-b3-traceid": {"TestB3TraceId"},
+		"x-foo":        {"TestFoo"},
+	}
 
 	// Create A Mock Kafka SyncProducer
 	mockSyncProducer := producertesting.NewMockSyncProducer()
@@ -78,7 +84,7 @@ func TestProduceKafkaMessage(t *testing.T) {
 	producer := createTestProducer(t, brokers, config, mockSyncProducer)
 
 	// Perform The Test & Verify Results
-	err := producer.ProduceKafkaMessage(context.Background(), channelReference, bindingMessage)
+	err := producer.ProduceKafkaMessage(context.Background(), channelReference, bindingMessage, httpHeader)
 	assert.Nil(t, err)
 
 	// Verify Message Was Produced Correctly
@@ -99,6 +105,11 @@ func TestProduceKafkaMessage(t *testing.T) {
 	receivertesting.ValidateProducerMessageHeader(t, producerMessage.Headers, constants.CeKafkaHeaderKeySubject, receivertesting.EventSubject)
 	receivertesting.ValidateProducerMessageHeader(t, producerMessage.Headers, constants.CeKafkaHeaderKeyDataSchema, receivertesting.EventDataSchema)
 	receivertesting.ValidateProducerMessageHeader(t, producerMessage.Headers, constants.CeKafkaHeaderKeyPartitionKey, receivertesting.PartitionKey)
+	for headerKey, headerValues := range httpHeader {
+		for _, headerValue := range headerValues {
+			receivertesting.ValidateProducerMessageHeader(t, producerMessage.Headers, headerKey, headerValue)
+		}
+	}
 }
 
 // Test The Producer's SecretChanged Functionality
diff --git a/pkg/common/tracing/tracing_test.go b/pkg/common/tracing/tracing_test.go
index 5429ba162..3a7184a12 100644
--- a/pkg/common/tracing/tracing_test.go
+++ b/pkg/common/tracing/tracing_test.go
@@ -18,14 +18,16 @@ package tracing
 
 import (
 	"context"
+	"net/http"
 	"testing"
 
+	"github.com/Shopify/sarama"
 	protocolkafka "github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2"
-	logtesting "knative.dev/pkg/logging/testing"
-
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opencensus.io/trace"
 	"go.opencensus.io/trace/tracestate"
+	logtesting "knative.dev/pkg/logging/testing"
 )
 
 var (
@@ -102,3 +104,147 @@ func TestStartTraceFromMessage(t *testing.T) {
 	require.NotNil(t, ctx)
 	require.NotNil(t, span)
 }
+
+func TestConvertHttpHeaderToRecordHeaders(t *testing.T) {
+
+	key1 := "K1"
+	value1 := "V1"
+
+	key2 := "K2"
+	value2A := "V2A"
+	value2B := "V2B"
+
+	testCases := []struct {
+		name   string
+		header http.Header
+		want   []sarama.RecordHeader
+	}{
+		{
+			name:   "nil http header",
+			header: nil,
+			want:   []sarama.RecordHeader{},
+		},
+		{
+			name:   "empty http header",
+			header: http.Header{},
+			want:   []sarama.RecordHeader{},
+		},
+		{
+			name:   "single http header",
+			header: http.Header{key1: []string{value1}},
+			want:   []sarama.RecordHeader{{Key: []byte(key1), Value: []byte(value1)}},
+		},
+		{
+			name:   "multiple http header",
+			header: http.Header{key1: []string{value1}, key2: []string{value2A, value2B}},
+			want: []sarama.RecordHeader{
+				{Key: []byte(key1), Value: []byte(value1)},
+				{Key: []byte(key2), Value: []byte(value2A)},
+				{Key: []byte(key2), Value: []byte(value2B)},
+			},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			actual := ConvertHttpHeaderToRecordHeaders(testCase.header)
+			assert.Equal(t, testCase.want, actual)
+		})
+	}
+}
+
+func TestConvertRecordHeadersToHttpHeader(t *testing.T) {
+
+	key1 := "K1"
+	value1 := "V1"
+
+	key2 := "K2"
+	value2A := "V2A"
+	value2B := "V2B"
+
+	testCases := []struct {
+		name          string
+		recordHeaders []*sarama.RecordHeader
+		want          http.Header
+	}{
+		{
+			name:          "no headers",
+			recordHeaders: []*sarama.RecordHeader{},
+			want:          http.Header{},
+		},
+		{
+			name: "single header",
+			recordHeaders: []*sarama.RecordHeader{
+				{Key: []byte(key1), Value: []byte(value1)},
+			},
+			want: http.Header{key1: []string{value1}},
+		},
+		{
+			name: "multiple headers",
+			recordHeaders: []*sarama.RecordHeader{
+				{Key: []byte(key1), Value: []byte(value1)},
+				{Key: []byte(key2), Value: []byte(value2A)},
+				{Key: []byte(key2), Value: []byte(value2B)},
+			},
+			want: http.Header{
+				key1: []string{value1},
+				key2: []string{value2A, value2B},
+			},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			actual := ConvertRecordHeadersToHttpHeader(testCase.recordHeaders)
+			assert.Equal(t, testCase.want, actual)
+		})
+	}
+}
+
+func TestFilterCeRecordHeaders(t *testing.T) {
+
+	nonCeRecordHeader1 := &sarama.RecordHeader{Key: []byte("non_ce_key_1"), Value: []byte("non_ce_value_1")}
+	nonCeRecordHeader2 := &sarama.RecordHeader{Key: []byte("non_ce_key_2"), Value: []byte("non_ce_value_2")}
+
+	ceRecordHeader1 := &sarama.RecordHeader{Key: []byte("ce_key_1"), Value: []byte("ce_value_1")}
+	ceRecordHeader2 := &sarama.RecordHeader{Key: []byte("ce_key_2"), Value: []byte("ce_value_2")}
+
+	testCases := []struct {
+		name          string
+		recordHeaders []*sarama.RecordHeader
+		want          []*sarama.RecordHeader
+	}{
+		{
+			name:          "nil headers",
+			recordHeaders: nil,
+			want:          []*sarama.RecordHeader{},
+		},
+		{
+			name:          "empty headers",
+			recordHeaders: []*sarama.RecordHeader{},
+			want:          []*sarama.RecordHeader{},
+		},
+		{
+			name:          "no filtered headers",
+			recordHeaders: []*sarama.RecordHeader{nonCeRecordHeader1, nonCeRecordHeader2},
+			want:          []*sarama.RecordHeader{nonCeRecordHeader1, nonCeRecordHeader2},
+		},
+		{
+			name:          "some filtered headers",
+			recordHeaders: []*sarama.RecordHeader{ceRecordHeader1, nonCeRecordHeader1, ceRecordHeader2, nonCeRecordHeader2},
+			want:          []*sarama.RecordHeader{nonCeRecordHeader1, nonCeRecordHeader2},
+		},
+		{
+			name:          "all filtered headers",
+			recordHeaders: []*sarama.RecordHeader{ceRecordHeader1, ceRecordHeader2},
+			want:          []*sarama.RecordHeader{},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			actual := FilterCeRecordHeaders(testCase.recordHeaders)
+			assert.Equal(t, testCase.want, actual)
+		})
+	}
+}