// 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"]
Use the Kafka Bridge to produce messages to a Kafka topic in JSON format by using the topics endpoint.

You can produce messages to topics in JSON format by using the xref:send[topics] endpoint.
You can specify destination partitions for messages in the request body.
The xref:sendtopartition[partitions] endpoint provides an alternative method for specifying a single destination partition for all messages as a path parameter.

In this procedure, messages are produced to a topic called `bridge-quickstart-topic`.

.Prerequisites

* The Kafka cluster has a topic with three partitions.
+
--
You can use the `kafka-topics.sh` utility to create topics.

.Example topic creation with three partitions
[source,shell,subs=attributes+]
----
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic bridge-quickstart-topic --partitions 3 --replication-factor 1
----

.Verifying the topic was created
[source,shell,subs=attributes+]
----
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic bridge-quickstart-topic
----
--

NOTE: If you deployed Strimzi on Kubernetes, you can create a topic using the `KafkaTopic` custom resource.

.Procedure

. 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

* xref:send[POST /topics/{topicname}]
* xref:sendtopartition[POST /topics/{topicname}/partitions/{partitionid}]
