Repository: cloudevents/sdk-go
Pull request #388 (2020-03-16 17:55:10)

Adding trace support back in.
Signed-off-by: Scott Nichols <snichols@vmware.com>


Patch:
diff --git a/alias.go b/alias.go
index 2bf5df6..b4a4837 100644
--- a/alias.go
+++ b/alias.go
@@ -79,15 +79,16 @@ var (
 	// Client Creation
 
 	NewClient             = client.New
+	NewClientObserved     = client.NewObserved
 	NewDefaultClient      = client.NewDefault
 	NewHTTPReceiveHandler = client.NewHTTPReceiveHandler
 
 	// Client Options
 
-	WithEventDefaulter      = client.WithEventDefaulter
-	WithUUIDs               = client.WithUUIDs
-	WithTimeNow             = client.WithTimeNow
-	WithoutTracePropagation = client.WithoutTracePropagation
+	WithEventDefaulter   = client.WithEventDefaulter
+	WithUUIDs            = client.WithUUIDs
+	WithTimeNow          = client.WithTimeNow
+	WithTracePropagation = client.WithTracePropagation()
 
 	// Event Creation
 
diff --git a/cmd/samples/http/receiver-traced/main.go b/cmd/samples/http/receiver-traced/main.go
new file mode 100644
index 0000000..ebe58fe
--- /dev/null
+++ b/cmd/samples/http/receiver-traced/main.go
@@ -0,0 +1,33 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"github.com/cloudevents/sdk-go/pkg/client"
+	"log"
+
+	cloudevents "github.com/cloudevents/sdk-go"
+)
+
+func main() {
+	ctx := context.Background()
+	p, err := cloudevents.NewHTTP()
+	if err != nil {
+		log.Fatalf("failed to create protocol: %s", err.Error())
+	}
+
+	c, err := cloudevents.NewClientObserved(p, cloudevents.WithTracePropagation)
+	if err != nil {
+		log.Fatalf("failed to create client, %v", err)
+	}
+
+	log.Printf("will listen on :8080\n")
+	log.Fatalf("failed to start receiver: %s", c.StartReceiver(ctx, receive))
+}
+
+func receive(ctx context.Context, e cloudevents.Event) {
+	ctx, span := client.TraceSpan(ctx, e)
+	defer span.End()
+
+	fmt.Printf("%s", e)
+}
diff --git a/cmd/samples/httpb/responder/main.go b/cmd/samples/httpb/responder/main.go
index 97f2e3b..bfbd256 100644
--- a/cmd/samples/httpb/responder/main.go
+++ b/cmd/samples/httpb/responder/main.go
@@ -36,7 +36,6 @@ func main() {
 	c, err := cloudevents.NewClient(p,
 		cloudevents.WithUUIDs(),
 		cloudevents.WithTimeNow(),
-		cloudevents.WithoutTracePropagation(),
 	)
 	if err != nil {
 		log.Fatalf("failed to create client: %s", err.Error())
diff --git a/cmd/samples/kafka/receiver/main.go b/cmd/samples/kafka/receiver/main.go
new file mode 100644
index 0000000..30cdf0b
--- /dev/null
+++ b/cmd/samples/kafka/receiver/main.go
@@ -0,0 +1,38 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	"github.com/Shopify/sarama"
+
+	cloudevents "github.com/cloudevents/sdk-go"
+	"github.com/cloudevents/sdk-go/pkg/transport/kafka_sarama"
+)
+
+func main() {
+	saramaConfig := sarama.NewConfig()
+
+	receiver, err := kafka_sarama.NewConsumer([]string{"127.0.0.1:9092"}, saramaConfig, "test-group-id", "test-topic")
+	if err != nil {
+		log.Fatalf("failed to create protocol: %s", err.Error())
+	}
+
+	c, err := cloudevents.NewClient(receiver)
+	if err != nil {
+		log.Fatalf("failed to create client, %v", err)
+	}
+
+	log.Printf("will listen consuming topic test-topic\n")
+	err = c.StartReceiver(context.TODO(), receive)
+	if err != nil {
+		log.Fatalf("failed to start receiver: %s", err)
+	} else {
+		log.Printf("receiver stopped\n")
+	}
+}
+
+func receive(ctx context.Context, event cloudevents.Event) {
+	fmt.Printf("%s", event)
+}
diff --git a/cmd/samples/kafka/sender/main.go b/cmd/samples/kafka/sender/main.go
new file mode 100644
index 0000000..c35cd5b
--- /dev/null
+++ b/cmd/samples/kafka/sender/main.go
@@ -0,0 +1,42 @@
+package main
+
+import (
+	"context"
+	"log"
+
+	"github.com/Shopify/sarama"
+
+	cloudevents "github.com/cloudevents/sdk-go"
+	"github.com/cloudevents/sdk-go/pkg/transport/kafka_sarama"
+)
+
+func main() {
+	saramaConfig := sarama.NewConfig()
+
+	sender, err := kafka_sarama.NewSender([]string{"127.0.0.1:9092"}, saramaConfig, "test-topic")
+	if err != nil {
+		log.Fatalf("failed to create protocol: %s", err.Error())
+	}
+
+	c, err := cloudevents.NewClient(sender, cloudevents.WithTimeNow(), cloudevents.WithUUIDs())
+	if err != nil {
+		log.Fatalf("failed to create client, %v", err)
+	}
+
+	for i := 0; i < 10; i++ {
+		e := cloudevents.NewEvent()
+		e.SetType("com.cloudevents.sample.sent")
+		e.SetSource("https://github.com/cloudevents/sdk-go/cmd/samples/httpb/requester")
+		_ = e.SetData(cloudevents.ApplicationJSON, map[string]interface{}{
+			"id":      i,
+			"message": "Hello, World!",
+		})
+
+		err := c.Send(context.Background(), e)
+		if err != nil {
+			log.Printf("failed to send: %v", err)
+		} else {
+			log.Printf("sent: %d", i)
+		}
+	}
+}
diff --git a/cmd/samples/kafka/sender_and_receiver/main.go b/cmd/samples/kafka/sender_and_receiver/main.go
new file mode 100644
index 0000000..b8d48f6
--- /dev/null
+++ b/cmd/samples/kafka/sender_and_receiver/main.go
@@ -0,0 +1,60 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	"github.com/Shopify/sarama"
+
+	cloudevents "github.com/cloudevents/sdk-go"
+	"github.com/cloudevents/sdk-go/pkg/transport/kafka_sarama"
+)
+
+func main() {
+	saramaConfig := sarama.NewConfig()
+
+	// With NewProtocol you can use the same client both to send and receive.
+	protocol, err := kafka_sarama.NewProtocol([]string{"127.0.0.1:9092"}, saramaConfig, "send-test-topic", "receive-test-topic")
+	if err != nil {
+		log.Fatalf("failed to create protocol: %s", err.Error())
+	}
+
+	c, err := cloudevents.NewClient(protocol, cloudevents.WithTimeNow(), cloudevents.WithUUIDs())
+	if err != nil {
+		log.Fatalf("failed to create client, %v", err)
+	}
+
+	// Start the receiver
+	go func() {
+		log.Printf("will listen consuming topic test-topic\n")
+		err = c.StartReceiver(context.TODO(), receive)
+		if err != nil {
+			log.Fatalf("failed to start receiver: %s", err)
+		} else {
+			log.Printf("receiver stopped\n")
+		}
+	}()
+
+	// Start sending the events
+	for i := 0; i < 10; i++ {
+		e := cloudevents.NewEvent()
+		e.SetType("com.cloudevents.sample.sent")
+		e.SetSource("https://github.com/cloudevents/sdk-go/cmd/samples/httpb/requester")
+		_ = e.SetData(cloudevents.ApplicationJSON, map[string]interface{}{
+			"id":      i,
+			"message": "Hello, World!",
+		})
+
+		err := c.Send(context.Background(), e)
+		if err != nil {
+			log.Printf("failed to send: %v", err)
+		} else {
+			log.Printf("sent: %d", i)
+		}
+	}
+}
+
+func receive(ctx context.Context, event cloudevents.Event) {
+	fmt.Printf("%s", event)
+}
diff --git a/pkg/client/client_observed.go b/pkg/client/client_observed.go
index 99445e3..200ca64 100644
--- a/pkg/client/client_observed.go
+++ b/pkg/client/client_observed.go
@@ -3,6 +3,7 @@ package client
 import (
 	"context"
 	"github.com/cloudevents/sdk-go/pkg/event"
+	"github.com/cloudevents/sdk-go/pkg/extensions"
 	"github.com/cloudevents/sdk-go/pkg/observability"
 	"go.opencensus.io/trace"
 )
@@ -15,34 +16,43 @@ func NewObserved(protocol interface{}, opts ...Option) (Client, error) {
 		return nil, err
 	}
 
-	return &obsClient{client: client}, nil
+	c := &obsClient{client: client}
+
+	if err := c.applyOptions(opts...); err != nil {
+		return nil, err
+	}
+	return c, nil
 }
 
 type obsClient struct {
 	client Client
 
-	disableTracePropagation bool // TODO?
+	addTracing bool
 }
 
-//
-//func (c *ceClient) applyOptions(opts ...Option) error {
-//	for _, fn := range opts {
-//		if err := fn(c); err != nil {
-//			return err
-//		}
-//	}
-//	return nil
-//}
+func (c *obsClient) applyOptions(opts ...Option) error {
+	for _, fn := range opts {
+		if err := fn(c); err != nil {
+			return err
+		}
+	}
+	return nil
+}
 
 // Send transmits the provided event on a preconfigured Protocol. Send returns
 // an error if there was an an issue validating the outbound event or the
 // transport returns an error.
 func (c *obsClient) Send(ctx context.Context, e event.Event) error {
 	ctx, r := observability.NewReporter(ctx, reportSend)
-	ctx, span := trace.StartSpan(ctx, clientSpanName, trace.WithSpanKind(trace.SpanKindClient))
+	ctx, span := trace.StartSpan(ctx, observability.ClientSpanName, trace.WithSpanKind(trace.SpanKindClient))
 	defer span.End()
 	if span.IsRecordingEvents() {
-		span.AddAttributes(eventTraceAttributes(e.Context)...)
+		span.AddAttributes(EventTraceAttributes(&e)...)
+	}
+
+	if c.addTracing {
+		e.Context = e.Context.Clone()
+		extensions.FromSpanContext(span.SpanContext()).AddTracingAttributes(&e)
 	}
 
 	err := c.client.Send(ctx, e)
@@ -57,10 +67,10 @@ func (c *obsClient) Send(ctx context.Context, e event.Event) error {
 
 func (c *obsClient) Request(ctx context.Context, e event.Event) (*event.Event, error) {
 	ctx, r := observability.NewReporter(ctx, reportRequest)
-	ctx, span := trace.StartSpan(ctx, clientSpanName, trace.WithSpanKind(trace.SpanKindClient))
+	ctx, span := trace.StartSpan(ctx, observability.ClientSpanName, trace.WithSpanKind(trace.SpanKindClient))
 	defer span.End()
 	if span.IsRecordingEvents() {
-		span.AddAttributes(eventTraceAttributes(e.Context)...)
+		span.AddAttributes(EventTraceAttributes(&e)...)
 	}
 
 	resp, err := c.client.Request(ctx, e)
diff --git a/pkg/client/invoker.go b/pkg/client/invoker.go
index 07cb866..542e874 100644
--- a/pkg/client/invoker.go
+++ b/pkg/client/invoker.go
@@ -31,7 +31,7 @@ func newReceiveInvoker(fn interface{}, fns ...EventDefaulter) (Invoker, error) {
 
 type receiveInvoker struct {
 	fn                *receiverFn
-	eventDefaulterFns []EventDefaulter // TODO: set these.
+	eventDefaulterFns []EventDefaulter
 }
 
 func (r *receiveInvoker) Invoke(ctx context.Context, m binding.Message, respFn transport.ResponseFn) (err error) {
diff --git a/pkg/client/observability.go b/pkg/client/observability.go
index cc360df..ccad6bf 100644
--- a/pkg/client/observability.go
+++ b/pkg/client/observability.go
@@ -1,7 +1,9 @@
 package client
 
 import (
+	"context"
 	"github.com/cloudevents/sdk-go/pkg/event"
+	"github.com/cloudevents/sdk-go/pkg/extensions"
 	"github.com/cloudevents/sdk-go/pkg/observability"
 	"go.opencensus.io/stats"
 	"go.opencensus.io/stats/view"
@@ -31,8 +33,6 @@ type observed int32
 var _ observability.Observable = observed(0)
 
 const (
-	clientSpanName = "cloudevents.client"
-
 	specversionAttr     = "cloudevents.specversion"
 	typeAttr            = "cloudevents.type"
 	sourceAttr          = "cloudevents.source"
@@ -63,17 +63,32 @@ func (o observed) LatencyMs() *stats.Float64Measure {
 	return LatencyMs
 }
 
-func eventTraceAttributes(e event.EventContextReader) []trace.Attribute {
+func EventTraceAttributes(e event.EventReader) []trace.Attribute {
 	as := []trace.Attribute{
-		trace.StringAttribute(specversionAttr, e.GetSpecVersion()),
-		trace.StringAttribute(typeAttr, e.GetType()),
-		trace.StringAttribute(sourceAttr, e.GetSource()),
+		trace.StringAttribute(specversionAttr, e.SpecVersion()),
+		trace.StringAttribute(typeAttr, e.Type()),
+		trace.StringAttribute(sourceAttr, e.Source()),
 	}
-	if sub := e.GetSubject(); sub != "" {
+	if sub := e.Subject(); sub != "" {
 		as = append(as, trace.StringAttribute(subjectAttr, sub))
 	}
-	if dct := e.GetDataContentType(); dct != "" {
+	if dct := e.DataContentType(); dct != "" {
 		as = append(as, trace.StringAttribute(datacontenttypeAttr, dct))
 	}
 	return as
 }
+
+// TraceSpan returns context and trace.Span based on event. Caller must call span.End()
+func TraceSpan(ctx context.Context, e event.Event) (context.Context, *trace.Span) {
+	var span *trace.Span
+	if ext, ok := extensions.GetDistributedTracingExtension(e); ok {
+		ctx, span = ext.StartChildSpan(ctx, observability.ClientSpanName, trace.WithSpanKind(trace.SpanKindServer))
+	}
+	if span == nil {
+		ctx, span = trace.StartSpan(ctx, observability.ClientSpanName, trace.WithSpanKind(trace.SpanKindServer))
+	}
+	if span.IsRecordingEvents() {
+		span.AddAttributes(EventTraceAttributes(&e)...)
+	}
+	return ctx, span
+}
diff --git a/pkg/client/options.go b/pkg/client/options.go
index 4a5539f..110fca4 100644
--- a/pkg/client/options.go
+++ b/pkg/client/options.go
@@ -6,29 +6,35 @@ import (
 )
 
 // Option is the function signature required to be considered an client.Option.
-type Option func(*ceClient) error
+type Option func(interface{}) error
 
 // WithEventDefaulter adds an event defaulter to the end of the defaulter chain.
 func WithEventDefaulter(fn EventDefaulter) Option {
-	return func(c *ceClient) error {
-		if fn == nil {
-			return fmt.Errorf("client option was given an nil event defaulter")
+	return func(i interface{}) error {
+		if c, ok := i.(*ceClient); ok {
+			if fn == nil {
+				return fmt.Errorf("client option was given an nil event defaulter")
+			}
+			c.eventDefaulterFns = append(c.eventDefaulterFns, fn)
 		}
-		c.eventDefaulterFns = append(c.eventDefaulterFns, fn)
 		return nil
 	}
 }
 
 func WithForceBinary() Option {
-	return func(c *ceClient) error {
-		c.outboundContextDecorators = append(c.outboundContextDecorators, binding.WithForceBinary)
+	return func(i interface{}) error {
+		if c, ok := i.(*ceClient); ok {
+			c.outboundContextDecorators = append(c.outboundContextDecorators, binding.WithForceBinary)
+		}
 		return nil
 	}
 }
 
 func WithForceStructured() Option {
-	return func(c *ceClient) error {
-		c.outboundContextDecorators = append(c.outboundContextDecorators, binding.WithForceStructured)
+	return func(i interface{}) error {
+		if c, ok := i.(*ceClient); ok {
+			c.outboundContextDecorators = append(c.outboundContextDecorators, binding.WithForceStructured)
+		}
 		return nil
 	}
 }
@@ -36,8 +42,10 @@ func WithForceStructured() Option {
 // WithUUIDs adds DefaultIDToUUIDIfNotSet event defaulter to the end of the
 // defaulter chain.
 func WithUUIDs() Option {
-	return func(c *ceClient) error {
-		c.eventDefaulterFns = append(c.eventDefaulterFns, DefaultIDToUUIDIfNotSet)
+	return func(i interface{}) error {
+		if c, ok := i.(*ceClient); ok {
+			c.eventDefaulterFns = append(c.eventDefaulterFns, DefaultIDToUUIDIfNotSet)
+		}
 		return nil
 	}
 }
@@ -45,17 +53,21 @@ func WithUUIDs() Option {
 // WithTimeNow adds DefaultTimeToNowIfNotSet event defaulter to the end of the
 // defaulter chain.
 func WithTimeNow() Option {
-	return func(c *ceClient) error {
-		c.eventDefaulterFns = append(c.eventDefaulterFns, DefaultTimeToNowIfNotSet)
+	return func(i interface{}) error {
+		if c, ok := i.(*ceClient); ok {
+			c.eventDefaulterFns = append(c.eventDefaulterFns, DefaultTimeToNowIfNotSet)
+		}
 		return nil
 	}
 }
 
-// WithoutTracePropagation disables automatic trace propagation via
-// the distributed tracing extension.
-func WithoutTracePropagation() Option {
-	return func(c *ceClient) error {
-		//c.disableTracePropagation = true
+// WithTracePropagation enables trace propagation via the distributed tracing
+// extension.
+func WithTracePropagation() Option {
+	return func(i interface{}) error {
+		if c, ok := i.(*obsClient); ok {
+			c.addTracing = true
+		}
 		return nil
 	}
 }
diff --git a/pkg/extensions/distributed_tracing_extension.go b/pkg/extensions/distributed_tracing_extension.go
index 37968fa..5989ff1 100644
--- a/pkg/extensions/distributed_tracing_extension.go
+++ b/pkg/extensions/distributed_tracing_extension.go
@@ -19,11 +19,6 @@ const (
 	TraceStateExtension  = "tracestate"
 )
 
-// EventTracer interface allows setting extension for cloudevents context.
-type EventTracer interface {
-	SetExtension(k string, v interface{}) error
-}
-
 // DistributedTracingExtension represents the extension for cloudevents context
 type DistributedTracingExtension struct {
 	TraceParent string `json:"traceparent"`
@@ -31,7 +26,7 @@ type DistributedTracingExtension struct {
 }
 
 // AddTracingAttributes adds the tracing attributes traceparent and tracestate to the cloudevents context
-func (d DistributedTracingExtension) AddTracingAttributes(ec EventTracer) error {
+func (d DistributedTracingExtension) AddTracingAttributes(e event.EventWriter) {
 	if d.TraceParent != "" {
 		value := reflect.ValueOf(d)
 		typeOf := value.Type()
@@ -42,12 +37,9 @@ func (d DistributedTracingExtension) AddTracingAttributes(ec EventTracer) error
 			if k == TraceStateExtension && v == "" {
 				continue
 			}
-			if err := ec.SetExtension(k, v); err != nil {
-				return err
-			}
+			e.SetExtension(k, v)
 		}
 	}
-	return nil
 }
 
 func GetDistributedTracingExtension(event event.Event) (DistributedTracingExtension, bool) {
diff --git a/pkg/observability/keys.go b/pkg/observability/keys.go
index f032b10..afadddc 100644
--- a/pkg/observability/keys.go
+++ b/pkg/observability/keys.go
@@ -12,6 +12,9 @@ var (
 )
 
 const (
+	// ClientSpanName is the key used to start spans from the client.
+	ClientSpanName = "cloudevents.client"
+
 	// ResultError is a shared result tag value for error.
 	ResultError = "error"
 	// ResultOK is a shared result tag value for success.
diff --git a/pkg/transport/kafka_sarama/option.go b/pkg/transport/kafka_sarama/option.go
index dc3219f..0b3b359 100644
--- a/pkg/transport/kafka_sarama/option.go
+++ b/pkg/transport/kafka_sarama/option.go
@@ -1,6 +1,10 @@
 package kafka_sarama
 
-import "github.com/cloudevents/sdk-go/pkg/binding"
+import (
+	"context"
+
+	"github.com/cloudevents/sdk-go/pkg/binding"
+)
 
 // kafka_sarama.Sender options
 type SenderOptionFunc func(sender *Sender)
@@ -11,3 +15,18 @@ func WithTransformer(transformer binding.TransformerFactory) SenderOptionFunc {
 		sender.transformers = append(sender.transformers, transformer)
 	}
 }
+
+// kafka_sarama.Protocol options
+type ProtocolOptionFunc func(protocol *Protocol)
+
+func WithReceiverGroupId(groupId string) ProtocolOptionFunc {
+	return func(protocol *Protocol) {
+		protocol.receiverGroupId = groupId
+	}
+}
+
+func WithSenderContextDecorators(decorator func(context.Context) context.Context) ProtocolOptionFunc {
+	return func(protocol *Protocol) {
+		protocol.SenderContextDecorators = append(protocol.SenderContextDecorators, decorator)
+	}
+}
diff --git a/pkg/transport/kafka_sarama/protocol.go b/pkg/transport/kafka_sarama/protocol.go
new file mode 100644
index 0000000..6975df8
--- /dev/null
+++ b/pkg/transport/kafka_sarama/protocol.go
@@ -0,0 +1,126 @@
+package kafka_sarama
+
+import (
+	"context"
+	"errors"
+	"sync"
+
+	"github.com/Shopify/sarama"
+
+	"github.com/cloudevents/sdk-go/pkg/binding"
+	cecontext "github.com/cloudevents/sdk-go/pkg/context"
+	"github.com/cloudevents/sdk-go/pkg/transport"
+)
+
+const (
+	defaultGroupId = "cloudevents-sdk-go"
+)
+
+type Protocol struct {
+	// Kafka
+	Client sarama.Client
+
+	// Sender
+	Sender *Sender
+
+	// Sender options
+	SenderContextDecorators []func(context.Context) context.Context
+	senderTransformers      binding.TransformerFactories
+	senderTopic             string
+
+	// Consumer
+	Consumer    *Consumer
+	consumerMux sync.Mutex
+
+	// Consumer options
+	receiverTopic   string
+	receiverGroupId string
+}
+
+// NewProtocol creates a new kafka transport.
+func NewProtocol(brokers []string, saramaConfig *sarama.Config, sendToTopic string, receiveFromTopic string, opts ...ProtocolOptionFunc) (*Protocol, error) {
+	client, err := sarama.NewClient(brokers, saramaConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewProtocolFromClient(client, sendToTopic, receiveFromTopic, opts...)
+}
+
+// NewProtocolFromClient creates a new kafka transport starting from a sarama.Client
+func NewProtocolFromClient(client sarama.Client, sendToTopic string, receiveFromTopic string, opts ...ProtocolOptionFunc) (*Protocol, error) {
+	p := &Protocol{
+		Client:                  client,
+		SenderContextDecorators: make([]func(context.Context) context.Context, 0),
+		receiverGroupId:         defaultGroupId,
+		senderTopic:             sendToTopic,
+		receiverTopic:           receiveFromTopic,
+	}
+
+	var err error
+	if err = p.applyOptions(opts...); err != nil {
+		return nil, err
+	}
+
+	if p.senderTopic == "" {
+		return nil, errors.New("you didn't specify the topic to send to")
+	}
+	p.Sender, err = NewSenderFromClient(p.Client, p.senderTopic)
+	if err != nil {
+		return nil, err
+	}
+
+	if p.receiverTopic == "" {
+		return nil, errors.New("you didn't specify the topic to receive from")
+	}
+	p.Consumer = NewConsumerFromClient(p.Client, p.receiverGroupId, p.receiverTopic)
+
+	return p, nil
+}
+
+func (p *Protocol) applyOptions(opts ...ProtocolOptionFunc) error {
+	for _, fn := range opts {
+		fn(p)
+	}
+	return nil
+}
+
+// StartReceiver implements Protocol.StartReceiver
+// NOTE: This is a blocking call.
+func (p *Protocol) OpenInbound(ctx context.Context) error {
+	p.consumerMux.Lock()
+	defer p.consumerMux.Unlock()
+
+	logger := cecontext.LoggerFrom(ctx)
+	logger.Infof("Starting consumer group to topic {} and group id {}", p.receiverTopic, p.receiverGroupId)
+
+	return p.Consumer.OpenInbound(ctx)
+}
+
+// HasTracePropagation implements Protocol.HasTracePropagation
+func (p *Protocol) HasTracePropagation() bool {
+	return false
+}
+
+func (p *Protocol) Send(ctx context.Context, in binding.Message) error {
+	for _, f := range p.SenderContextDecorators {
+		ctx = f(ctx)
+	}
+	return p.Sender.Send(ctx, in)
+}
+
+func (p *Protocol) Receive(ctx context.Context) (binding.Message, error) {
+	return p.Consumer.Receive(ctx)
+}
+
+func (p *Protocol) Close(ctx context.Context) error {
+	if err := p.Consumer.Close(ctx); err != nil {
+		return err
+	}
+	return p.Sender.Close(ctx)
+}
+
+// Kafka protocol implements Sender, Receiver
+var _ transport.Sender = (*Protocol)(nil)
+var _ transport.Receiver = (*Protocol)(nil)
+var _ transport.Closer = (*Protocol)(nil)
diff --git a/pkg/transport/kafka_sarama/receiver.go b/pkg/transport/kafka_sarama/receiver.go
index 53c0a91..f8f8fd4 100644
--- a/pkg/transport/kafka_sarama/receiver.go
+++ b/pkg/transport/kafka_sarama/receiver.go
@@ -3,10 +3,12 @@ package kafka_sarama
 import (
 	"context"
 	"io"
+	"sync"
 
 	"github.com/Shopify/sarama"
 
 	"github.com/cloudevents/sdk-go/pkg/binding"
+	"github.com/cloudevents/sdk-go/pkg/transport"
 )
 
 type msgErr struct {
@@ -17,22 +19,16 @@ type msgErr struct {
 // Receiver which implements sarama.ConsumerGroupHandler
 // After the first invocation of Receiver.Receive(), the sarama.ConsumerGroup is created and started.
 type Receiver struct {
+	once     sync.Once
 	incoming chan msgErr
-
-	client              sarama.Client
-	topic               string
-	groupId             string
-	saramaConsumerGroup sarama.ConsumerGroup
 }
 
 // NewReceiver creates a Receiver which implements sarama.ConsumerGroupHandler
+// The sarama.ConsumerGroup must be started invoking. If you need a Receiver which also manage the ConsumerGroup, use NewConsumer
 // After the first invocation of Receiver.Receive(), the sarama.ConsumerGroup is created and started.
-func NewReceiver(client sarama.Client, groupId string, topic string) *Receiver {
+func NewReceiver() *Receiver {
 	return &Receiver{
 		incoming: make(chan msgErr),
-		client:   client,
-		groupId:  groupId,
-		topic:    topic,
 	}
 }
 
@@ -41,6 +37,9 @@ func (r *Receiver) Setup(sess sarama.ConsumerGroupSession) error {
 }
 
 func (r *Receiver) Cleanup(sarama.ConsumerGroupSession) error {
+	r.once.Do(func() {
+		close(r.incoming)
+	})
 	return nil
 }
 
@@ -56,21 +55,6 @@ func (r *Receiver) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 }
 
 func (r *Receiver) Receive(ctx context.Context) (binding.Message, error) {
-	// Consumer Group not started!
-	if r.saramaConsumerGroup == nil {
-		cg, err := sarama.NewConsumerGroupFromClient(r.groupId, r.client)
-		if err != nil {
-			return nil, err
-		}
-		r.saramaConsumerGroup = cg
-
-		go func() {
-			if err := cg.Consume(ctx, []string{r.topic}, r); err != nil {
-				r.incoming <- msgErr{err: err}
-			}
-		}()
-	}
-
 	msgErr, ok := <-r.incoming
 	if !ok {
 		return nil, io.EOF
@@ -78,9 +62,68 @@ func (r *Receiver) Receive(ctx context.Context) (binding.Message, error) {
 	return msgErr.msg, msgErr.err
 }
 
-func (r *Receiver) Close(ctx context.Context) error {
-	if r.saramaConsumerGroup != nil {
-		return r.saramaConsumerGroup.Close()
+var _ transport.Receiver = (*Receiver)(nil)
+
+type Consumer struct {
+	Receiver
+
+	client              sarama.Client
+	topic               string
+	groupId             string
+	saramaConsumerGroup sarama.ConsumerGroup
+}
+
+func NewConsumer(brokers []string, saramaConfig *sarama.Config, groupId string, topic string) (*Consumer, error) {
+	client, err := sarama.NewClient(brokers, saramaConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewConsumerFromClient(client, groupId, topic), nil
+}
+
+func NewConsumerFromClient(client sarama.Client, groupId string, topic string) *Consumer {
+	return &Consumer{
+		Receiver: Receiver{
+			incoming: make(chan msgErr),
+		},
+		client:  client,
+		topic:   topic,
+		groupId: groupId,
+	}
+}
+
+func (c *Consumer) OpenInbound(ctx context.Context) (err error) {
+	cg, err := sarama.NewConsumerGroupFromClient(c.groupId, c.client)
+	if err != nil {
+		return
+	}
+	c.saramaConsumerGroup = cg
+
+	errCh := make(chan error)
+
+	go func(errs chan error) {
+		errs <- cg.Consume(context.Background(), []string{c.topic}, c)
+	}(errCh)
+
+	for {
+		select {
+		case <-ctx.Done():
+			err = c.Close(context.TODO())
+			return
+		case err = <-errCh:
+			return
+		}
+	}
+}
+
+func (c *Consumer) Close(ctx context.Context) error {
+	if c.saramaConsumerGroup != nil {
+		if err := c.saramaConsumerGroup.Close(); err != nil {
+			return err
+		}
 	}
 	return nil
 }
+
+var _ transport.Opener = (*Consumer)(nil)
diff --git a/pkg/transport/kafka_sarama/sender.go b/pkg/transport/kafka_sarama/sender.go
index a575991..44f0028 100644
--- a/pkg/transport/kafka_sarama/sender.go
+++ b/pkg/transport/kafka_sarama/sender.go
@@ -8,7 +8,7 @@ import (
 	"github.com/cloudevents/sdk-go/pkg/binding"
 )
 
-// Sender implements binding.Sender that sends messages to a specific topic using sarama.SyncProducer
+// Sender implements binding.Sender that sends messages to a specific receiverTopic using sarama.SyncProducer
 type Sender struct {
 	topic        string
 	syncProducer sarama.SyncProducer
@@ -16,8 +16,18 @@ type Sender struct {
 	transformers binding.TransformerFactories
 }
 
-// Returns a binding.Sender that sends messages to a specific topic using sarama.SyncProducer
-func NewSender(client sarama.Client, topic string, options ...SenderOptionFunc) (*Sender, error) {
+// Returns a binding.Sender that sends messages to a specific receiverTopic using sarama.SyncProducer
+func NewSender(brokers []string, saramaConfig *sarama.Config, topic string, options ...SenderOptionFunc) (*Sender, error) {
+	client, err := sarama.NewClient(brokers, saramaConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewSenderFromClient(client, topic, options...)
+}
+
+// Returns a binding.Sender that sends messages to a specific receiverTopic using sarama.SyncProducer
+func NewSenderFromClient(client sarama.Client, topic string, options ...SenderOptionFunc) (*Sender, error) {
 	producer, err := sarama.NewSyncProducerFromClient(client)
 	if err != nil {
 		return nil, err


Test patch:
diff --git a/pkg/client/client_test.go b/pkg/client/client_test.go
index 673878d..80ec599 100644
--- a/pkg/client/client_test.go
+++ b/pkg/client/client_test.go
@@ -43,7 +43,7 @@ func simpleBinaryClient(target string) client.Client {
 		return nil
 	}
 
-	c, err := client.New(p, client.WithoutTracePropagation(), client.WithForceBinary())
+	c, err := client.New(p, client.WithForceBinary())
 	if err != nil {
 		return nil
 	}
@@ -57,7 +57,7 @@ func simpleTracingBinaryClient(target string) client.Client {
 		return nil
 	}
 
-	c, err := client.New(p)
+	c, err := client.NewObserved(p, client.WithTracePropagation())
 	if err != nil {
 		return nil
 	}
@@ -71,7 +71,7 @@ func simpleStructuredClient(target string) client.Client {
 		return nil
 	}
 
-	c, err := client.New(p, client.WithoutTracePropagation(), client.WithForceStructured())
+	c, err := client.New(p, client.WithForceStructured())
 	if err != nil {
 		return nil
 	}
@@ -187,8 +187,6 @@ func TestClientSend(t *testing.T) {
 }
 
 func TestTracingClientSend(t *testing.T) {
-	t.Skip("skipping tracing tests for now, need to rework this for sdk v2")
-
 	now := time.Now()
 
 	testCases := map[string]struct {
@@ -475,8 +473,6 @@ func TestClientReceive(t *testing.T) {
 }
 
 func TestTracedClientReceive(t *testing.T) {
-	t.Skip("TODO: need to re-add tracedClient features into httpb")
-
 	now := time.Now()
 
 	testCases := map[string]struct {
@@ -518,9 +514,11 @@ func TestTracedClientReceive(t *testing.T) {
 
 			ctx, cancel := context.WithCancel(context.TODO())
 			go func() {
-				err = c.StartReceiver(ctx, func(ctx context.Context, event event.Event) (*event.Event, transport.Result) {
+				err = c.StartReceiver(ctx, func(ctx context.Context, e event.Event) (*event.Event, transport.Result) {
 					go func() {
-						spanContexts <- trace.FromContext(ctx).SpanContext()
+						_, span := client.TraceSpan(ctx, e)
+						defer span.End()
+						spanContexts <- span.SpanContext()
 					}()
 					return nil, nil
 				})
@@ -531,10 +529,10 @@ func TestTracedClientReceive(t *testing.T) {
 			time.Sleep(5 * time.Millisecond) // let the server start
 
 			target := fmt.Sprintf("http://localhost:%d", p.GetPort())
-			client := simpleBinaryClient(target)
+			sender := simpleTracingBinaryClient(target)
 
 			ctx, span := trace.StartSpan(context.TODO(), "test-span")
-			err = client.Send(ctx, tc.event)
+			err = sender.Send(ctx, tc.event)
 			span.End()
 
 			if err != nil {
diff --git a/pkg/client/test/test.go b/pkg/client/test/test.go
new file mode 100644
index 0000000..d8b3b4d
--- /dev/null
+++ b/pkg/client/test/test.go
@@ -0,0 +1,56 @@
+package test
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/cloudevents/sdk-go/pkg/client"
+	"github.com/cloudevents/sdk-go/pkg/event"
+	"github.com/cloudevents/sdk-go/pkg/transport"
+)
+
+// SendReceive does client.Send(in), then it receives the message using client.StartReceiver() and executes outAssert
+// Halt test on error.
+func SendReceive(t *testing.T, protocolFactory func() interface{}, in event.Event, outAssert func(e event.Event), opts ...client.Option) {
+	t.Helper()
+	protocol := protocolFactory()
+	c, err := client.New(protocol, opts...)
+	require.NoError(t, err)
+	wg := sync.WaitGroup{}
+	wg.Add(2)
+
+	go func() {
+		ctx, cancel := context.WithCancel(context.TODO())
+		ch := make(chan event.Event)
+		defer func(channel chan event.Event) {
+			cancel()
+			close(channel)
+			wg.Done()
+		}(ch)
+		go func(channel chan event.Event) {
+			err := c.StartReceiver(ctx, func(e event.Event) {
+				channel <- e
+			})
+			if err != nil {
+				require.NoError(t, err)
+			}
+		}(ch)
+		e := <-ch
+		outAssert(e)
+	}()
+
+	go func() {
+		defer wg.Done()
+		err := c.Send(context.Background(), in)
+		require.NoError(t, err)
+	}()
+
+	wg.Wait()
+
+	if closer, ok := protocol.(transport.Closer); ok {
+		require.NoError(t, closer.Close(context.TODO()))
+	}
+}
diff --git a/pkg/extensions/distributed_tracing_extension_test.go b/pkg/extensions/distributed_tracing_extension_test.go
index 4350d56..30365a6 100644
--- a/pkg/extensions/distributed_tracing_extension_test.go
+++ b/pkg/extensions/distributed_tracing_extension_test.go
@@ -177,16 +177,13 @@ func testAddTracingAttributesFunc(t *testing.T, st extensions.DistributedTracing
 	var e event.Event
 	switch ces {
 	case "EventContextV1":
-		ectx := ecv.context.(event.EventContextV1).AsV1()
-		st.AddTracingAttributes(ectx)
-		e = event.Event{Context: ectx}
+		e = event.Event{Context: ecv.context.(event.EventContextV1).AsV1()}
 		e.SetData(event.ApplicationJSON, &Data{Message: "Hello world"})
 	case "EventContextV03":
-		ectx := ecv.context.(event.EventContextV03).AsV03()
-		st.AddTracingAttributes(ectx)
-		e = event.Event{Context: ectx}
+		e = event.Event{Context: ecv.context.(event.EventContextV03).AsV03()}
 		e.SetData(event.ApplicationJSON, &Data{Message: "Hello world"})
 	}
+	st.AddTracingAttributes(&e)
 	got := e.Extensions()
 
 	if diff := cmp.Diff(ecv.want, got); diff != "" {
diff --git a/pkg/transport/kafka_sarama/message_benchmark_test.go b/pkg/transport/kafka_sarama/message_benchmark_test.go
index f343528..9435cc1 100644
--- a/pkg/transport/kafka_sarama/message_benchmark_test.go
+++ b/pkg/transport/kafka_sarama/message_benchmark_test.go
@@ -37,7 +37,7 @@ var (
 			"ce_specversion":     "1.0",
 			"ce_dataschema":      test.Schema.String(),
 			"ce_datacontenttype": "text/json",
-			"ce_subject":         "topic",
+			"ce_subject":         "receiverTopic",
 			"ce_exta":            "someext",
 		}),
 	}
diff --git a/test/integration/http/direct.go b/test/integration/http/direct.go
index 80caed2..55ce22b 100644
--- a/test/integration/http/direct.go
+++ b/test/integration/http/direct.go
@@ -46,7 +46,6 @@ func ClientDirect(t *testing.T, tc DirectTapTest, copts ...client.Option) {
 	tap.handler = protocol
 
 	copts = append(copts, cloudevents.WithEventDefaulter(AlwaysThen(tc.now)))
-	copts = append(copts, cloudevents.WithoutTracePropagation())
 
 	ce, err := cloudevents.NewClient(protocol, copts...)
 	if err != nil {
diff --git a/test/integration/http/loopback.go b/test/integration/http/loopback.go
index d9cc2fe..0c8a1a2 100644
--- a/test/integration/http/loopback.go
+++ b/test/integration/http/loopback.go
@@ -58,7 +58,6 @@ func ClientLoopback(t *testing.T, tc TapTest, copts ...client.Option) {
 	tap.handler = protocol
 
 	copts = append(copts, cloudevents.WithEventDefaulter(AlwaysThen(tc.now)))
-	copts = append(copts, cloudevents.WithoutTracePropagation())
 
 	ce, err := cloudevents.NewClient(protocol, copts...)
 	if err != nil {
diff --git a/test/integration/http/middleware.go b/test/integration/http/middleware.go
index dcf81c0..4bcbc06 100644
--- a/test/integration/http/middleware.go
+++ b/test/integration/http/middleware.go
@@ -29,7 +29,6 @@ func ClientMiddleware(t *testing.T, tc TapTest, copts ...client.Option) {
 	tap.handler = protocol
 
 	copts = append(copts, cloudevents.WithEventDefaulter(AlwaysThen(tc.now)))
-	copts = append(copts, cloudevents.WithoutTracePropagation())
 
 	ce, err := cloudevents.NewClient(protocol, copts...)
 	if err != nil {
diff --git a/test/integration/kafka_sarama/kafka_test.go b/test/integration/kafka_sarama/kafka_test.go
new file mode 100644
index 0000000..4565673
--- /dev/null
+++ b/test/integration/kafka_sarama/kafka_test.go
@@ -0,0 +1,66 @@
+package kafka_sarama_binding
+
+import (
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/Shopify/sarama"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+
+	bindingtest "github.com/cloudevents/sdk-go/pkg/binding/test"
+	clienttest "github.com/cloudevents/sdk-go/pkg/client/test"
+	"github.com/cloudevents/sdk-go/pkg/event"
+	"github.com/cloudevents/sdk-go/pkg/transport/kafka_sarama"
+)
+
+const (
+	TEST_GROUP_ID = "test_group_id"
+)
+
+func TestSendEvent(t *testing.T) {
+	bindingtest.EachEvent(t, bindingtest.Events(), func(t *testing.T, eventIn event.Event) {
+		eventIn = bindingtest.ExToStr(t, eventIn)
+		clienttest.SendReceive(t, func() interface{} {
+			return testSenderReceiver(t)
+		}, eventIn, func(e event.Event) {
+			bindingtest.AssertEventEquals(t, eventIn, bindingtest.ExToStr(t, e))
+		})
+	})
+}
+
+// To start a local environment for testing:
+// docker run --rm --net=host -e ADV_HOST=localhost lensesio/fast-data-dev
+func testClient(t testing.TB) sarama.Client {
+	t.Helper()
+	s := os.Getenv("TEST_KAFKA_BOOTSTRAP_SERVER")
+	if s == "" {
+		s = "localhost:9092"
+	}
+
+	config := sarama.NewConfig()
+	config.Version = sarama.V2_0_0_0
+	config.Producer.Return.Successes = true
+	config.Producer.Return.Errors = true
+	config.Consumer.Offsets.Initial = sarama.OffsetOldest
+	client, err := sarama.NewClient(strings.Split(s, ","), config)
+	if err != nil {
+		t.Skipf("Cannot create sarama client to servers [%s]: %v", s, err)
+	}
+
+	return client
+}
+
+func testSenderReceiver(t testing.TB) *kafka_sarama.Protocol {
+	client := testClient(t)
+
+	topicName := "test-ce-client-" + uuid.New().String()
+	options := []kafka_sarama.ProtocolOptionFunc{
+		kafka_sarama.WithReceiverGroupId(TEST_GROUP_ID),
+	}
+	p, err := kafka_sarama.NewProtocolFromClient(client, topicName, topicName, options...)
+	require.NoError(t, err)
+
+	return p
+}
diff --git a/test/integration/kafka_sarama_binding/kafka_test.go b/test/integration/kafka_sarama_binding/kafka_test.go
index 7ba93b8..2ef37de 100644
--- a/test/integration/kafka_sarama_binding/kafka_test.go
+++ b/test/integration/kafka_sarama_binding/kafka_test.go
@@ -53,6 +53,8 @@ func TestSendBinaryMessageToBinary(t *testing.T) {
 	})
 }
 
+// To start a local environment for testing:
+// docker run --rm --net=host -e ADV_HOST=localhost lensesio/fast-data-dev
 func testClient(t testing.TB) sarama.Client {
 	t.Helper()
 	s := os.Getenv("TEST_KAFKA_BOOTSTRAP_SERVER")
@@ -77,10 +79,14 @@ func testSenderReceiver(t testing.TB, options ...kafka_sarama.SenderOptionFunc)
 	client := testClient(t)
 
 	topicName := "test-ce-client-" + uuid.New().String()
-	r := kafka_sarama.NewReceiver(client, TEST_GROUP_ID, topicName)
-	s, err := kafka_sarama.NewSender(client, topicName, options...)
+	r := kafka_sarama.NewConsumerFromClient(client, TEST_GROUP_ID, topicName)
+	s, err := kafka_sarama.NewSenderFromClient(client, topicName, options...)
 	require.NoError(t, err)
 
+	go func() {
+		require.NoError(t, r.OpenInbound(context.TODO()))
+	}()
+
 	return func() {
 		err = r.Close(context.TODO())
 		require.NoError(t, err)