Skip to content
This repository has been archived by the owner on Jun 4, 2021. It is now read-only.

Fixed kafka source after update to eventing #1223

Merged
merged 1 commit into from
May 13, 2020
Merged

Fixed kafka source after update to eventing #1223

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion kafka/source/pkg/reconciler/source/resources/receive_adapter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ func MakeReceiveAdapter(args *ReceiveAdapterArgs) *v1.Deployment {
Name: "KAFKA_NET_TLS_ENABLE",
Value: strconv.FormatBool(args.Source.Spec.Net.TLS.Enable),
}, {
Name: "SINK_URI",
Name: "K_SINK",
Copy link
Member

@matzew matzew May 13, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

that is the SinkBinding feature ...

Do we no longer need the old, just the new ? or was that all changed ?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can't really say why this env changed, i just switched from adapter to adapter/v2 and this happened :)

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

right -- just to elaborate on that, the K_SINK is a var in the adapter.EnvConfig and it gets resolved to the sink, so that's why it still properly outputs in the logging.

Value: args.SinkURI,
}, {
Name: "NAME",
Expand Down
2 changes: 1 addition & 1 deletion test/e2e/helpers/kafka_helper.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ func MustPublishKafkaMessage(client *testlib.Client, bootstrapServer string, top
}
args = append(args, "/etc/mounted/payload")

client.T.Logf("Running kafkacat %v", args)
client.T.Logf("Running kafkacat %s", strings.Join(args, " "))

pod := corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Expand Down
2 changes: 0 additions & 2 deletions test/e2e/kafka_binding_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,6 @@ import (
)

func testKafkaBinding(t *testing.T, messageKey string, messageHeaders map[string]string, messagePayload string, expectedCheckInLog string) {
t.Skipf("failing e2e tests. TODO: Fix this.")

client := lib.Setup(t, true)

kafkaTopicName := uuid.New().String()
Expand Down
2 changes: 0 additions & 2 deletions test/e2e/kafka_source_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,8 +70,6 @@ func testKafkaSource(t *testing.T, messageKey string, messageHeaders map[string]
}

func TestKafkaSource(t *testing.T) {
t.Skipf("failing e2e tests. TODO: Fix this.")

tests := map[string]struct {
messageKey string
messageHeaders map[string]string
Expand Down