amqp producer writing data into only one partition

[iamrsaravana]

Please use this to only for bug reports. For questions or when you need help, you can use the GitHub Discussions, our #strimzi Slack channel or out user mailing list.

Describe the bug
A clear and concise description of what the bug is.

i am using topic configuration:
topic.yaml

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
name: my-topic-excavator
labels:
strimzi.io/cluster: my-cluster
spec:
partitions: 3
replicas: 3
config:
retention.ms: 7200000
segment.bytes: 1073741824

Connect.yaml

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
name: my-connect-cluster-excavator
annotations:
strimzi.io/use-connector-resources: "true"
spec:
image: 10.128.0.6:5000/my-connector-crusher1
replicas: 1
bootstrapServers: my-cluster-kafka-bootstrap:9092
config:
group.id: connect-cluster-new-excavator
#key.converter: org.apache.kafka.connect.storage.StringConverter
#key.converter: org.apache.kafka.connect.json.JsonConverter
#value.converter: org.apache.kafka.connect.json.JsonConverter
#key.converter.schemas.enable: false
#value.converter.schemas.enable: false
offset.storage.topic: connect-cluster-offsets-excavator
config.storage.topic: connect-cluster-configs-excavator
status.storage.topic: connect-cluster-status-excavator
config.storage.replication.factor: 3
offset.storage.replication.factor: 3
status.storage.replication.factor: 3

AMQP source connector:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
name: camelamqpsourceconnector-excavator
labels:
strimzi.io/cluster: my-connect-cluster-excavator
spec:
class: org.apache.camel.kafkaconnector.amqp.CamelAmqpSourceConnector
tasksMax: 1

topics: mytopic

config:
camel.component.amqp.includeAmqpAnnotations: true
camel.component.amqp.connectionFactory: '#class:org.apache.qpid.jms.JmsConnectionFactory'
camel.component.amqp.connectionFactory.remoteURI: amqp://10.97.217.229:15672
camel.component.amqp.username: consumer@HONO
camel.component.amqp.password: verysecret
camel.component.amqp.testConnectionOnStartup: true
camel.source.path.destinationType: queue
camel.source.path.destinationName: telemetry/1_TENANT
topics: my-topic-excavator

the above configuration writing only to one parition(0) only all the data. we are not able to see data in parition 1 or 2

./bin/kafka-console-consumer.sh --bootstrap-server 10.104.7.152:9092 --topic my-topic-excavator --partition 1

To Reproduce
Steps to reproduce the behavior:

1. Go to '...'
2. Create Custom Resource '....'
3. Run command '....'
4. See error

Expected behavior
A clear and concise description of what you expected to happen.

Environment (please complete the following information):

• Strimzi version: [e.g. main, 0.22.1]
• Installation method: [e.g. YAML files, Helm chart, OperatorHub.io]
• Kubernetes cluster: [e.g. Kubernetes 1.20, OpenShift 4.7]
• Infrastructure: [e.g. Amazon EKS, Minikube]

YAML files and logs

Attach or copy paste the custom resources you used to deploy the Kafka cluster and the relevant YAMLs created by the Cluster Operator.
Attach or copy and paste also the relevant logs.

To easily collect all YAMLs and logs, you can use our report script which will automatically collect all files and prepare a ZIP archive which can be easily attached to this issue.
The usage of this script is:
./report.sh [--namespace <string>] [--cluster <string>]

Additional context
Add any other context about the problem here.

[scholzj]

Could you please use code formatting to make the YAML example readable? The way you pasted them makes it impossible to see the structure. As for the actual issue, I have no idea how the connector or connect decide where to send the message. A lot will depend on the connector what key does it give to the message etc. If it gives it always the same key for example, the messages will always end up in the same partition -> this is an important design feature of Kafka.

[iamrsaravana]

topics:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
name: my-topic-excavator
labels:
strimzi.io/cluster: my-cluster
spec:
partitions: 3
replicas: 3
config:
retention.ms: 7200000
segment.bytes: 1073741824

Connect.yaml:
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
name: my-connect-cluster-excavator
annotations:
strimzi.io/use-connector-resources: "true"
spec:
image: 10.128.0.6:5000/my-connector-crusher1
replicas: 3
bootstrapServers: my-cluster-kafka-bootstrap:9092
config:
group.id: connect-cluster-new-excavator
#key.converter: org.apache.kafka.connect.storage.StringConverter
#key.converter: org.apache.kafka.connect.json.JsonConverter
#value.converter: org.apache.kafka.connect.json.JsonConverter
#key.converter.schemas.enable: false
#value.converter.schemas.enable: false
offset.storage.topic: connect-cluster-offsets-excavator
config.storage.topic: connect-cluster-configs-excavator
status.storage.topic: connect-cluster-status-excavator
config.storage.replication.factor: 3
offset.storage.replication.factor: 3
status.storage.replication.factor: 3

Amqp connector.yaml

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
name: camelamqpsourceconnector-excavator
labels:
strimzi.io/cluster: my-connect-cluster-excavator
spec:
class: org.apache.camel.kafkaconnector.amqp.CamelAmqpSourceConnector
tasksMax: 3

topics: mytopic

config:
camel.component.amqp.includeAmqpAnnotations: true
camel.component.amqp.connectionFactory: '#class:org.apache.qpid.jms.JmsConnectionFactory'
camel.component.amqp.connectionFactory.remoteURI: amqp://10.97.217.229:15672
camel.component.amqp.username: consumer@HONO
camel.component.amqp.password: verysecret
camel.component.amqp.testConnectionOnStartup: true
camel.source.path.destinationType: queue
camel.source.path.destinationName: telemetry/1_TENANT
topics: my-topic-excavator

we are sending data from eclipse hono(source is amqp connector) which is sending data to kafka topics.

[iamrsaravana]

This is my source data coming from amqp adapter:

12:37:26.263 [vert.x-eventloop-thread-0] INFO org.eclipse.hono.cli.app.Receiver - ... with application properties: {orig_adapter=hono-mqtt, qos=1, device_id=Device58, orig_address=telemetry}
12:37:26.334 [vert.x-eventloop-thread-0] INFO org.eclipse.hono.cli.app.Receiver - received telemetry message [device: Device141, content-type: application/octet-stream]: {'Header': [{'ID': 'Device141', 'Type': 'PCR', 'Mvmt': 'rotary', 'Fuel': 'petrol', 'SNo': 'G000123', 'IMEI': 'GNPS235765443', 'ICCID': 'IC200009', 'CCode': 'LnT', 'Ver': 'V01.02.03', 'Volt': 43.44, 'Batt': 78, 'GSM': 55, 'GPS': 94}], 'Data': [{'RTC': '2021-06-07 12:37:26.332050', 'Lat': 15.748980619733661, 'Long': 16.846907143768806, 'M1': 3, 'M2': 4, 'Vr': 105.15834457250416, 'Vy': 108.20212453242456, 'Vb': 99.83829063848094, 'Lr': 99.84210155660374, 'Ly': 107.70750971052773, 'Lb': 102.69634486548325, 'Freq': 18.76426097683381, 'KVA': 17.192209900380192, 'PF': 19.817413120788498, 'KW': 17.58793039571805, 'KWH': 14.55737573267546, 'FwRPM': 14.963128297644774, 'RDT': 12.75247041373487, 'Pr': 14.460799929391857, 'Tmp': 12.10577175880044, 'TW': 13.124193966339611, 'Spd': 19.6006698337198, 'Flow': 19.584672576123513, 'Err': 14.894160675373746}]}
12:37:26.334 [vert.x-eventloop-thread-0] INFO org.eclipse.hono.cli.app.Receiver - ... with application properties: {orig_adapter=hono-mqtt, qos=1, device_id=Device141, orig_address=telemetry}
12:37:28.104 [vert.x-eventloop-thread-0] INFO org.eclipse.hono.cli.app.Receiver - received telemetry message [device: Device90, content-type: application/octet-stream]: {'Header': [{'ID': 'Device90', 'Type': 'PCR', 'Mvmt': 'rotary', 'Fuel': 'petrol', 'SNo': 'G000123', 'IMEI': 'GNPS235765443', 'ICCID': 'IC200009', 'CCode': 'LnT', 'Ver': 'V01.02.03', 'Volt': 43.44, 'Batt': 78, 'GSM': 55, 'GPS': 94}], 'Data': [{'RTC': '2021-06-07 12:37:28.101188', 'Lat': 15.924656878796068, 'Long': 15.940207989667996, 'M1': 1, 'M2': 1, 'Vr': 97.33936020681939, 'Vy': 105.44657795748525, 'Vb': 104.13395147555633, 'Lr': 99.35924673827162, 'Ly': 100.03500985417092, 'Lb': 100.34324464300983, 'Freq': 15.766395214968046, 'KVA': 16.83792201166522, 'PF': 13.639878533265474, 'KW': 15.394780263047826, 'KWH': 19.51453654304123, 'FwRPM': 18.003753882840634, 'RDT': 18.30292822506553, 'Pr': 19.826933849935674, 'Tmp': 16.294594981621426, 'TW': 12.52062946241633, 'Spd': 17.019226246284077, 'Flow': 19.933962508425065, 'Err': 13.874184254183916}]}
12:37:28.104 [vert.x-eventloop-thread-0] INFO org.eclipse.hono.cli.app.Receiver - ... with application properties: {orig_adapter=hono-mqtt, qos=1, device_id=Device90, orig_address=telemetry}

[iamrsaravana]

how do we change incoming key values from camel AMQP source connector?