// Module included in the following assemblies:
//
// assembly-kafka-bridge-quickstart.adoc

[id='proc-producing-messages-from-bridge-topics-partitions-{context}']
= Producing messages to topics and partitions

[role="_abstract"]
After you exposed the service of your Kafka Bridge deployment, you can produce messages to topics in JSON format by using the link:https://strimzi.io/docs/bridge/latest/#_send[topics^] endpoint. You can specify destination partitions for messages in the request body, as shown here. The link:https://strimzi.io/docs/bridge/latest/#_sendtopartition[partitions^] endpoint provides an alternative method for specifying a single destination partition for all messages as a path parameter.

.Procedure

. In a text editor, create a YAML definition for a Kafka topic with three partitions.
+
[source,yaml,subs=attributes+]
----
apiVersion: {KafkaTopicApiVersion}
kind: KafkaTopic
metadata:
  name: bridge-quickstart-topic
  labels:
    strimzi.io/cluster: <kafka-cluster-name> <1>
spec:
  partitions: 3 <2>
  replicas: 1
  config:
    retention.ms: 7200000
    segment.bytes: 1073741824
----
<1> The name of the Kafka cluster in which the Kafka Bridge is deployed.
<2> The number of partitions for the topic.

. Save the file to the `examples/topic` directory as `bridge-quickstart-topic.yaml`.

. Create the topic in your Kubernetes cluster:
+
[source,shell,subs=attributes+]
----
kubectl apply -f examples/topic/bridge-quickstart-topic.yaml
----

. Using the Kafka Bridge, produce three messages to the topic you created:
+
[source,curl,subs=attributes+]
----
curl -X POST \
  http://localhost:8080/topics/bridge-quickstart-topic \
  -H 'content-type: application/vnd.kafka.json.v2+json' \
  -d '{
    "records": [
        {
            "key": "my-key",
            "value": "sales-lead-0001"
        },
        {
            "value": "sales-lead-0002",
            "partition": 2
        },
        {
            "value": "sales-lead-0003"
        }
    ]
}'
----
+
* `sales-lead-0001` is sent to a partition based on the hash of the key.
* `sales-lead-0002` is sent directly to partition 2.
* `sales-lead-0003` is sent to a partition in the `bridge-quickstart-topic` topic using a round-robin method.

. If the request is successful, the Kafka Bridge returns an `offsets` array, along with a `200` code and a `content-type` header of `application/vnd.kafka.v2+json`. For each message, the `offsets` array describes:
+
* The partition that the message was sent to
* The current message offset of the partition
+
.Example response

[source,json,subs=attributes+]
----
#...
{
  "offsets":[
    {
      "partition":0,
      "offset":0
    },
    {
      "partition":2,
      "offset":0
    },
    {
      "partition":0,
      "offset":1
    }
  ]
}
----

.Additional topic requests

Make other curl requests to find information on topics and partitions.

List topics::
+
--
[source,curl]
----
curl -X GET \
  http://localhost:8080/topics
----

.Example response
[source,json]
----
[
  "__strimzi_store_topic",
  "__strimzi-topic-operator-kstreams-topic-store-changelog",
  "bridge-quickstart-topic",
  "my-topic"
]
----
--

Get topic configuration and partition details::
+
--
[source,curl,subs=attributes+]
----
curl -X GET \
  http://localhost:8080/topics/bridge-quickstart-topic
----

.Example response
[source,json]
----
{
  "name": "bridge-quickstart-topic",
  "configs": {
    "compression.type": "producer",
    "leader.replication.throttled.replicas": "",
    "min.insync.replicas": "1",
    "message.downconversion.enable": "true",
    "segment.jitter.ms": "0",
    "cleanup.policy": "delete",
    "flush.ms": "9223372036854775807",
    "follower.replication.throttled.replicas": "",
    "segment.bytes": "1073741824",
    "retention.ms": "604800000",
    "flush.messages": "9223372036854775807",
    "message.format.version": "2.8-IV1",
    "max.compaction.lag.ms": "9223372036854775807",
    "file.delete.delay.ms": "60000",
    "max.message.bytes": "1048588",
    "min.compaction.lag.ms": "0",
    "message.timestamp.type": "CreateTime",
    "preallocate": "false",
    "index.interval.bytes": "4096",
    "min.cleanable.dirty.ratio": "0.5",
    "unclean.leader.election.enable": "false",
    "retention.bytes": "-1",
    "delete.retention.ms": "86400000",
    "segment.ms": "604800000",
    "message.timestamp.difference.max.ms": "9223372036854775807",
    "segment.index.bytes": "10485760"
  },
  "partitions": [
    {
      "partition": 0,
      "leader": 0,
      "replicas": [
        {
          "broker": 0,
          "leader": true,
          "in_sync": true
        },
        {
          "broker": 1,
          "leader": false,
          "in_sync": true
        },
        {
          "broker": 2,
          "leader": false,
          "in_sync": true
        }
      ]
    },
    {
      "partition": 1,
      "leader": 2,
      "replicas": [
        {
          "broker": 2,
          "leader": true,
          "in_sync": true
        },
        {
          "broker": 0,
          "leader": false,
          "in_sync": true
        },
        {
          "broker": 1,
          "leader": false,
          "in_sync": true
        }
      ]
    },
    {
      "partition": 2,
      "leader": 1,
      "replicas": [
        {
          "broker": 1,
          "leader": true,
          "in_sync": true
        },
        {
          "broker": 2,
          "leader": false,
          "in_sync": true
        },
        {
          "broker": 0,
          "leader": false,
          "in_sync": true
        }
      ]
    }
  ]
}
----
--

List the partitions of a specific topic::
+
--
[source,curl,subs=attributes+]
----
curl -X GET \
  http://localhost:8080/topics/bridge-quickstart-topic/partitions
----

.Example response
[source,json]
----
[
  {
    "partition": 0,
    "leader": 0,
    "replicas": [
      {
        "broker": 0,
        "leader": true,
        "in_sync": true
      },
      {
        "broker": 1,
        "leader": false,
        "in_sync": true
      },
      {
        "broker": 2,
        "leader": false,
        "in_sync": true
      }
    ]
  },
  {
    "partition": 1,
    "leader": 2,
    "replicas": [
      {
        "broker": 2,
        "leader": true,
        "in_sync": true
      },
      {
        "broker": 0,
        "leader": false,
        "in_sync": true
      },
      {
        "broker": 1,
        "leader": false,
        "in_sync": true
      }
    ]
  },
  {
    "partition": 2,
    "leader": 1,
    "replicas": [
      {
        "broker": 1,
        "leader": true,
        "in_sync": true
      },
      {
        "broker": 2,
        "leader": false,
        "in_sync": true
      },
      {
        "broker": 0,
        "leader": false,
        "in_sync": true
      }
    ]
  }
]
----
--

List the details of a specific topic partition::
+
--
[source,curl,subs=attributes+]
----
curl -X GET \
  http://localhost:8080/topics/bridge-quickstart-topic/partitions/0
----

.Example response
[source,json]
----
{
  "partition": 0,
  "leader": 0,
  "replicas": [
    {
      "broker": 0,
      "leader": true,
      "in_sync": true
    },
    {
      "broker": 1,
      "leader": false,
      "in_sync": true
    },
    {
      "broker": 2,
      "leader": false,
      "in_sync": true
    }
  ]
}
----
--

List the offsets of a specific topic partition::
+
--
[source,curl,subs=attributes+]
----
curl -X GET \
  http://localhost:8080/topics/bridge-quickstart-topic/partitions/0/offsets
----

.Example response
[source,json]
----
{
  "beginning_offset": 0,
  "end_offset": 1
}
----
--

.What to do next

After producing messages to topics and partitions, xref:proc-creating-kafka-bridge-consumer-{context}[create a Kafka Bridge consumer].

[role="_additional-resources"]
.Additional resources

* link:https://strimzi.io/docs/bridge/latest/#_send[POST /topics/{topicname}^] in the API reference documentation.
* link:https://strimzi.io/docs/bridge/latest/#_sendtopartition[POST /topics/{topicname}/partitions/{partitionid}^] in the API reference documentation.
