Add kafka source

[vrazdalovschi]

Closes #9

[snowplowcla]

Thanks for your pull request. Is this your first contribution to a Snowplow open source project? Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

:memo: Please visit https://docs.snowplowanalytics.com/docs/contributing/contributor-license-agreement/ to learn more and sign.

Once you've signed, please reply here (e.g. I signed it!) and we'll verify. Thanks.

[vrazdalovschi]

colmsnowplow, Let me know your thoughts about this PR. The code is based on the feat/kafka_source branch. I have added additional e2e tests and additional configuration parameters, fixed minor things. I'd be happy to move forward with all comments and progress on making it live ASAP.

[vrazdalovschi]

I signed it!

[snowplowcla]

Confirmed! @vrazdalovschi has signed the Contributor License Agreement. Thanks so much.

[colmsnowplow]

Thanks @vrazdalovschi!

It's turning out to be quite a busy week so please bear with me, I will take a look as soon as I can. :)

[vrazdalovschi]

@colmsnowplow, Hi. I have realized that I do not need the SnowBridge project in my stack right now, but I'd like to finish this PR, so there is no pushing to the timeline. Right now, I'm pretty busy with other personal things for the next ~10 days, but I'll try in the meantime to apply most of the fixes and take a look at the testing. If the test is time-consuming, then I might delay that part, and anyone could jump here :) Thank you for your review!

[colmsnowplow]

No problem, no rush. When you do get to it happy to discuss/offer whatever help I can. Hopefully we find some space to chip away at it our side too :)

[vrazdalovschi]

@colmsnowplow , hopefully, I have found some time and fixed the notes. The one area for improvement for me is testing. The e2e test is present in release_test/e2e_source_test.go. I haven't added mocked tests "integration" tests for kafka source. In case you want to include them, could you describe them?

1. To test happy flow, I have to mock the Consume func and to get that printed in STDOUT?
2. Anything else?

[colmsnowplow]

I might get to the in-code comments today but in case I don't, addressing the main Q:

I haven't added mocked tests "integration" tests for kafka source. In case you want to include them, could you describe them?

To test happy flow, I have to mock the Consume func and to get that printed in STDOUT?

"Integration test" normally describes something a bit different, so perhaps wrong word used. I just mean that we can use the existing local kafka cluster setup to "cheat" on unit tests and use those resources instead of mocks, to isolate the source function behaviour in a test.

Effectively this is covered in the end to end tests, but it's important to have because we need a way to easily isolate/replicate scenarios when we need to debug stuff etc. I'm especially cautious in this case since the original PR never reached a point where it proved itself to work via unit tests (so hesitant to merge to prod without a means of prodding at it basically).

The way this works is that when running tests locally, we can run with -short flag to skip the tests that rely on local resources like I propose here, or if we want to run those, we run make integration-up to set those up first. There's a kafka instance in the docker compose file that that command runs, so we'd basically be using that instead of mocking stuff.

The below is untested, but I've taken the parallel tests from pubsub source and will pseudocode out what I'd do now:

func TestKafkaSource_ReadAndReturnSuccessIntegration(t *testing.T) {
    if testing.Short() {
        t.Skip("skipping integration test")
    }
    assert := assert.New(t)

    // Create topic and subscription
    // topic, subscription := testutil.CreatePubSubTopicAndSubscription(t, "test-topic", "test-sub")
    // defer topic.Delete(context.Background())
    // defer subscription.Delete(context.Background())

// ^^ replace with what you've already done here: https://github.com/snowplow/snowbridge/blob/8dd6f57fbead54f194e054d29aa6e9550cc8eebe/release_test/e2e_source_test.go#L162C2-L178

    // Write to topic
    // testutil.WriteToPubSubTopic(t, topic, 10)

// ^^ replace with some code to write some messages to the kafka topic

// Then, below, just change the config stuff and the vars to be kafka source stuff, and it should be done.
    t.Setenv("SOURCE_NAME", "pubsub")
    t.Setenv("SOURCE_PUBSUB_SUBSCRIPTION_ID", "test-sub")
    t.Setenv("SOURCE_PUBSUB_PROJECT_ID", `project-test`)

    adaptedHandle := adapterGenerator(configFunction)

    pubsubSourceConfigPair := config.ConfigurationPair{Name: "pubsub", Handle: adaptedHandle}
    supportedSources := []config.ConfigurationPair{pubsubSourceConfigPair}

    pubsubConfig, err := config.NewConfig()
    assert.NotNil(pubsubConfig)
    assert.Nil(err)

    pubsubSource, err := sourceconfig.GetSource(pubsubConfig, supportedSources)

    assert.NotNil(pubsubSource)
    assert.Nil(err)
    assert.Equal("projects/project-test/subscriptions/test-sub", pubsubSource.GetID())

    output := testutil.ReadAndReturnMessages(pubsubSource, 5*time.Second, testutil.DefaultTestWriteBuilder, nil)
    assert.Equal(10, len(output))
    for _, message := range output {
        assert.Contains(string(message.Data), `message #`)
        assert.Nil(message.GetError())
    }
}

Like I mentioned - it's totally fine if you don't do this, you've already added the e2e stuff which is great. I'll just need to find space to add this kind of test (and then think about other test cases) before we merge. :)

[colmsnowplow]

Released in https://github.com/snowplow/snowbridge/pull/285 - thanks again @vrazdalovschi