Repository: knative-sandbox/eventing-kafka
Pull request #81 (2020-10-06 09:27:52)

Porting 4169 to here
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>


related: https://github.com/knative/eventing/pull/4168/files

/hold

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/81#" title="Author self-approved">matzew</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/master/OWNERS)~~ [matzew]

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":[]} -->
matzew: @aliok based on a slack convo, I port this here

Question is: backport to `eventing-contrib` 0.17 and/or 0.18 ? 


aliok: @matzew we need the same in KafkaSource webhook too: https://github.com/knative-sandbox/eventing-kafka/blob/3b547e91c5511bdda7a0f82aa2cabf713894c023/config/source/deployments/controller.yaml#L63

IMO we should backport this to 0.18 for sure. I also prefer this being backported to 0.17.

Patch:
diff --git a/config/channel/consolidated/deployments/webhook.yaml b/config/channel/consolidated/deployments/webhook.yaml
index 46b06882c..3c742308b 100644
--- a/config/channel/consolidated/deployments/webhook.yaml
+++ b/config/channel/consolidated/deployments/webhook.yaml
@@ -66,7 +66,9 @@ spec:
             httpHeaders:
             - name: k-kubelet-probe
               value: "webhook"
-        livenessProbe: *probe
+        livenessProbe:
+          <<: *probe
+          initialDelaySeconds: 20
 
       # Our webhook should gracefully terminate by lame ducking first, set this to a sufficiently
       # high value that we respect whatever value it has configured for the lame duck grace period.
diff --git a/config/source/deployments/controller.yaml b/config/source/deployments/controller.yaml
index cc9a9d306..8ee3fd5be 100644
--- a/config/source/deployments/controller.yaml
+++ b/config/source/deployments/controller.yaml
@@ -60,7 +60,9 @@ spec:
             httpHeaders:
             - name: k-kubelet-probe
               value: "webhook"
-        livenessProbe: *probe
+        livenessProbe:
+          <<: *probe
+          initialDelaySeconds: 20
 
       serviceAccount: kafka-controller-manager
       terminationGracePeriodSeconds: 10
diff --git a/contrib/kafka/source/PROJECT b/contrib/kafka/source/PROJECT
deleted file mode 100644
index ae98d46d9..000000000
--- a/contrib/kafka/source/PROJECT
+++ /dev/null
@@ -1,3 +0,0 @@
-version: "1"
-domain: eventing.knative.dev
-repo: knative.dev/eventing-kafka
diff --git a/contrib/kafka/source/README.md b/contrib/kafka/source/README.md
deleted file mode 100644
index d0e927efd..000000000
--- a/contrib/kafka/source/README.md
+++ /dev/null
@@ -1,55 +0,0 @@
-# Apache Kafka - Source
-
-The Apache Kafka Event source enables Knative Eventing integration with Apache
-Kafka. When a message is produced to Apache Kafka, the Apache Kafka Event Source
-will consume the produced message and post that message to the corresponding
-event sink.
-
-## Deployment steps
-
-1. Setup [Knative Eventing](../../DEVELOPMENT.md)
-1. If not done already, install an Apache Kafka cluster!
-
-   - For Kubernetes a simple installation is done using the
-     [Strimzi Kafka Operator](http://strimzi.io). Its installation
-     [guides](http://strimzi.io/quickstarts/) provide content for Kubernetes and
-     Openshift.
-
-   > Note: The `KafkaSource` is not limited to Apache Kafka installations on
-   > Kubernetes. It is also possible to use an off-cluster Apache Kafka
-   > installation.
-
-1. Now that Apache Kafka is installed, apply the `KafkaSource` config:
-
-   ```
-   ko apply -f config/
-   ```
-
-1. Create the `KafkaSource` custom objects, by configuring the required
-   `consumerGroup`, `bootstrapServers` and `topics` values on the CR file of
-   your source. Below is an example:
-
-   ```yaml
-   apiVersion: sources.knative.dev/v1alpha1
-   kind: KafkaSource
-   metadata:
-     name: kafka-source
-   spec:
-     consumerGroup: knative-group
-     # Broker URL. Replace this with the URLs for your kafka cluster,
-     # which is in the format of my-cluster-kafka-bootstrap.my-kafka-namespace:9092.
-     bootstrapServers:
-       - REPLACE_WITH_CLUSTER_URL
-     topics:
-       - knative-demo-topic
-     sink:
-       ref:
-         apiVersion: serving.knative.dev/v1alpha1
-         kind: Service
-         name: event-display
-   ```
-
-## Example
-
-A more detailed example of the `KafkaSource` can be found in the
-[Knative documentation](https://knative.dev/docs/eventing/samples/).
diff --git a/hack/release.sh b/hack/release.sh
index 5b33db51d..7973e699c 100755
--- a/hack/release.sh
+++ b/hack/release.sh
@@ -19,24 +19,36 @@
 
 source $(dirname $0)/../vendor/knative.dev/test-infra/scripts/release.sh
 
+export GO111MODULE=on
+
+# Yaml files to generate, and the source config dir for them.
+declare -A COMPONENTS
+COMPONENTS=(
+  ["channel-consolidated.yaml"]="config/channel/consolidated"
+  ["channel-distributed.yaml"]="config/channel/distributed"
+  ["source.yaml"]="config/source"
+)
+readonly COMPONENTS
+
 function build_release() {
    # Update release labels if this is a tagged release
   if [[ -n "${TAG}" ]]; then
-    echo "Tagged release, updating release labels to kafka.eventing.knative.dev/release: \"${TAG}\""
-    LABEL_YAML_CMD=(sed -e "s|kafka.eventing.knative.dev/release: devel|kafka.eventing.knative.dev/release: \"${TAG}\"|")
+    echo "Tagged release, updating release labels to contrib.eventing.knative.dev/release: \"${TAG}\""
+    LABEL_YAML_CMD=(sed -e "s|contrib.eventing.knative.dev/release: devel|contrib.eventing.knative.dev/release: \"${TAG}\"|")
   else
     echo "Untagged release, will NOT update release labels"
     LABEL_YAML_CMD=(cat)
   fi
 
-  local config="config"
-  echo "Building eventing-kafka"
-
-
-  export CGO_ENABLED=1
-  ko resolve ${KO_FLAGS} -f ${config}/ | "${LABEL_YAML_CMD[@]}" > "eventing-kafka-channel.yaml"
-  ARTIFACTS_TO_PUBLISH="eventing-kafka-channel.yaml"
+  local all_yamls=()
+  for yaml in "${!COMPONENTS[@]}"; do
+    local config="${COMPONENTS[${yaml}]}"
+    echo "Building Knative Eventing Contrib - ${config}"
+    # TODO(chizhg): reenable --strict mode after https://github.com/knative/test-infra/issues/1262 is fixed.
+    ko resolve ${KO_FLAGS} -f ${config}/ | "${LABEL_YAML_CMD[@]}" > ${yaml}
+    all_yamls+=(${yaml})
+  done
+  ARTIFACTS_TO_PUBLISH="${all_yamls[@]}"
 }
 
 main $@
-
diff --git a/pkg/source/adapter/adapter.go b/pkg/source/adapter/adapter.go
index 5fb6f2f3a..492f70e20 100644
--- a/pkg/source/adapter/adapter.go
+++ b/pkg/source/adapter/adapter.go
@@ -21,16 +21,17 @@ import (
 	"net/http"
 	"strings"
 
+	kafkasource "knative.dev/eventing-kafka/pkg/source"
+
 	"go.opencensus.io/trace"
 	"knative.dev/eventing/pkg/adapter/v2"
 	"knative.dev/eventing/pkg/kncloudevents"
-	"knative.dev/pkg/source"
+	pkgsource "knative.dev/pkg/source"
 
 	"context"
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
-	kafkabinding "knative.dev/eventing-kafka/contrib/kafka"
 	"knative.dev/eventing-kafka/pkg/common/consumer"
 	"knative.dev/pkg/logging"
 )
@@ -55,7 +56,7 @@ func NewEnvConfig() adapter.EnvConfigAccessor {
 type Adapter struct {
 	config            *adapterConfig
 	httpMessageSender *kncloudevents.HttpMessageSender
-	reporter          source.StatsReporter
+	reporter          pkgsource.StatsReporter
 	logger            *zap.SugaredLogger
 	keyTypeMapper     func([]byte) interface{}
 }
@@ -63,7 +64,7 @@ type Adapter struct {
 var _ adapter.MessageAdapter = (*Adapter)(nil)
 var _ adapter.MessageAdapterConstructor = NewAdapter
 
-func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, httpMessageSender *kncloudevents.HttpMessageSender, reporter source.StatsReporter) adapter.MessageAdapter {
+func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, httpMessageSender *kncloudevents.HttpMessageSender, reporter pkgsource.StatsReporter) adapter.MessageAdapter {
 	logger := logging.FromContext(ctx)
 	config := processed.(*adapterConfig)
 
@@ -90,7 +91,7 @@ func (a *Adapter) start(stopCh <-chan struct{}) error {
 	)
 
 	// init consumer group
-	addrs, config, err := kafkabinding.NewConfig(context.Background())
+	addrs, config, err := kafkasource.NewConfig(context.Background())
 	if err != nil {
 		return fmt.Errorf("failed to create the config: %w", err)
 	}
@@ -142,7 +143,7 @@ func (a *Adapter) Handle(ctx context.Context, msg *sarama.ConsumerMessage) (bool
 		return false, fmt.Errorf("%d %s", res.StatusCode, http.StatusText(res.StatusCode))
 	}
 
-	reportArgs := &source.ReportArgs{
+	reportArgs := &pkgsource.ReportArgs{
 		Namespace:     a.config.Namespace,
 		Name:          a.config.Name,
 		ResourceGroup: resourceGroup,
diff --git a/contrib/kafka/client.go b/pkg/source/client.go
similarity index 99%
rename from contrib/kafka/client.go
rename to pkg/source/client.go
index de0815be4..7e8795771 100644
--- a/contrib/kafka/client.go
+++ b/pkg/source/client.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package kafka
+package source
 
 import (
 	"context"


Test patch:
diff --git a/contrib/test/test_images/kafka-publisher/main.go b/contrib/test/test_images/kafka-publisher/main.go
index e8f5fc0da..3de776408 100644
--- a/contrib/test/test_images/kafka-publisher/main.go
+++ b/contrib/test/test_images/kafka-publisher/main.go
@@ -19,9 +19,10 @@ package main
 import (
 	"log"
 
+	"knative.dev/eventing-kafka/pkg/source"
+
 	"github.com/Shopify/sarama"
 	"github.com/kelseyhightower/envconfig"
-	"knative.dev/eventing-kafka/contrib/kafka"
 	"knative.dev/pkg/signals"
 )
 
@@ -44,7 +45,7 @@ func main() {
 	log.Printf("GOT: %v", s)
 
 	// Create a Kafka client from our Binding.
-	client, err := kafka.NewProducer(ctx)
+	client, err := source.NewProducer(ctx)
 	if err != nil {
 		log.Fatal(err.Error())
 	}
diff --git a/contrib/kafka/client_test.go b/pkg/source/client_test.go
similarity index 99%
rename from contrib/kafka/client_test.go
rename to pkg/source/client_test.go
index 96cd0d3bd..296a32359 100644
--- a/contrib/kafka/client_test.go
+++ b/pkg/source/client_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package kafka
+package source
 
 import (
 	"bytes"