<h2>Summary:</h2>
<p>I'm trying to use KNative eventing to expose a simple web application via a Kafka topic. The server should be able to handle multiple requests concurrently, but, unfortunately, it seems to handle them sequentially when I send them via Kafka. When making simple HTTP requests directly to the service, though, the concurrency is working fine.</p>
<h2>Setup:</h2>
<p>The setup only uses a <code>KafkaSource</code> which points to my KNative <code>Service</code>, and is using a Kafka instance deployed using the <code>bitnami/kafka</code> helm chart.</p>
<p>The version I'm using is <code>v1.7.1</code> for KNative serving and eventing, and <code>v1.7.0</code> for the Kafka eventing integration (from <code>knative-sandbox/eventing-kafka</code>).</p>
<h2>Code:</h2>
<p>The service I'm trying to deploy is a python FastAPI application that, upon receiving a request (with an ID of sorts), logs the received request, sleeps for 5 seconds, then returns a dummy message:</p>
<pre class="lang-py prettyprint-override"><code>import asyncio
from fastapi import FastAPI
from pydantic import BaseModel
import logging

logging.basicConfig(
    format=&quot;%(asctime)s %(levelname)-8s %(message)s&quot;,
    level=logging.DEBUG, datefmt=&quot;%Y-%m-%d %H:%M:%S&quot;,
)

app = FastAPI()

class Item(BaseModel):
    id: str

@app.post(&quot;/&quot;)
async def root(item: Item):
    logging.debug(f&quot;Request received with ID: {item.id}&quot;)
    await asyncio.sleep(5)
    logging.debug(f&quot;Request complete for ID: {item.id}&quot;)
    return {&quot;message&quot;: &quot;Hello World&quot;}
</code></pre>
<p>The app is served using uvicorn:</p>
<pre class="lang-bash prettyprint-override"><code>FROM python:3.9-slim

RUN pip install fastapi uvicorn

ADD main.py .

ENTRYPOINT uvicorn --host 0.0.0.0 --port 8877 main:app
</code></pre>
<p>The service deployment spec shows that I'm setting a <code>containerConcurrency</code> value that's greater than <code>1</code>:</p>
<pre class="lang-yaml prettyprint-override"><code>apiVersion: serving.knative.dev/v1
kind: Service
metadata:
  name: concurrency-test
spec:
  template:
    metadata:
      annotations:
        autoscaling.knative.dev/class: &quot;kpa.autoscaling.knative.dev&quot;
        autoscaling.knative.dev/metric: &quot;concurrency&quot;
        autoscaling.knative.dev/target: &quot;5&quot;
    spec:
      containerConcurrency: 5
      containers:
        - name: app
          image: dev.local/concurrency-test:latest
          imagePullPolicy: IfNotPresent
          ports:
            - containerPort: 8877
---
apiVersion: sources.knative.dev/v1beta1
kind: KafkaSource
metadata:
  name: concurrency-test
spec:
  consumerGroup: concurrency-test-group
  bootstrapServers:
  - kafka.default.svc.cluster.local:9092
  topics:
  - concurrency-test-requests
  sink:
    ref:
      apiVersion: serving.knative.dev/v1
      kind: Service
      name: concurrency-test
</code></pre>
<blockquote>
<p>Note: I also tried with <code>spec.consumers: 2</code> in the <code>KafkaSource</code> but the behavior was the same.</p>
</blockquote>
<h2>Logs:</h2>
<p>When sending two concurrent requests to the service directly with HTTP, the logs to look like this (both requests finish within 6 seconds, so concurrency is in effect):</p>
<pre><code>2022-09-12 02:14:36 DEBUG    Request received with ID: abc
2022-09-12 02:14:37 DEBUG    Request received with ID: def
2022-09-12 02:14:41 DEBUG    Request complete for ID: abc
INFO:     10.42.0.7:0 - &quot;POST / HTTP/1.1&quot; 200 OK
2022-09-12 02:14:42 DEBUG    Request complete for ID: def
INFO:     10.42.0.7:0 - &quot;POST / HTTP/1.1&quot; 200 OK
</code></pre>
<p>When sending requests via Kafka, though, the logs look like this (the requests are being processed one after the other):</p>
<pre><code>2022-09-12 02:14:55 DEBUG    Request received with ID: 111
2022-09-12 02:15:00 DEBUG    Request complete for ID: 111
INFO:     10.42.0.7:0 - &quot;POST / HTTP/1.1&quot; 200 OK
2022-09-12 02:15:00 DEBUG    Request received with ID: 222
2022-09-12 02:15:05 DEBUG    Request complete for ID: 222
INFO:     10.42.0.7:0 - &quot;POST / HTTP/1.1&quot; 200 OK
</code></pre>
<p>Please let me know if this sequential request handling is the expected behavior when using eventing with just a <code>KafkaSource</code>, and I hope there are ways for enabling concurrency in this setup.</p>

<p>Kafka provides ordering within a partition (the implementation is a distributed log). You may need to change the number of partitions on your Kafka topic to achieve higher parallelism; you may be able to also use the <a href="https://github.com/knative-sandbox/eventing-kafka/blob/main/pkg/apis/sources/v1beta1/kafka_types.go#L63" rel="nofollow noreferrer"><code>spec.consumers</code> value</a> to increase the throughput (untested).</p>
<p>I'd also encourage filing an issue <a href="https://github.com/knative-sandbox/eventing-kafka/issues/new" rel="nofollow noreferrer">in the <code>eventing-kafka</code> repo</a> with your problem and any additional knobs if there is other behavior you're looking for.</p>
