# Use Message Browser in Confluent Cloud
Use message browser, a built-in tool in the Confluent Cloud Console, to inspect,
produce, and download messages from Apache Kafka® topics in real time.
You can browse messages across all partitions, jump to specific offsets
or timestamps, produce new messages with or without schemas, and
download messages as JSON or CSV. Cloud Console validates
messages against their schema before you produce them to the topic.
You pay for data [consumed and produced](../billing/billing-dimensions.md#ingress-egress-billing) to your cluster by message browser.
Message browser uses the Kafka REST API to produce messages, which incurs an
additional ingress charge. For details, see [Ingress and egress](../billing/billing-dimensions.md#ingress-egress-billing).
**Considerations**
- By default, message browser displays newer data at the top of the messages pane.
- Filter across any of the available columns.
- You can only filter results that are displayed. To add more results to the current display,
increase the maximum results, manipulate the start time, or add partitions.
- Message browser displays messages that have associated schemas. Where multiple schema contexts are
available, you can select between contexts with a drop-down list. Message browser uses only [TopicNameStrategy](../sr/fundamentals/serdes-develop/index.md#sr-schemas-subject-name-strategy) schemas
to deserialize messages.
- Non-compacted topics show a tile for **Total messages** while compacted topics show a tile
for **Retained bytes**.
- When you select **By offset** with **Latest**, message browser initially shows the 10 latest messages of each partition, followed by any incoming messages.
- **Production in the last hour** tracks all message delivery attempts, regardless of success or failure.
## Inspect topics
Use message browser to inspect topic details and browse live streaming
messages. If you don’t already have data flowing through the topic, use the
[built-in produce new message to topic feature](#ccloud-msgs-produce-new) to send a message to the topic.
You can replay old data that was already produced as long as it hasn’t been compacted or deleted yet.
To browse a live stream of messages and inspect a topic:
1. Sign in to Confluent Cloud.
2. If you have more than one [environment](../security/access-control/hierarchy/cloud-environments.md#cloud-environments), select an environment.
3. Select a cluster.
4. Select **Topics** in the navigation menu. The Topics page appears.
5. Select the topic you want to inspect.
6. Select **Messages**. The **Messages** page displays incoming streaming messages for the topic.
### Why does message browser not show messages equal to maximum results?
Message browser might not always display results equal to the maximum results selected. This is by design and
can happen for various reasons based on your filtering selections and the ordering of data in the topic. For
example, your topic might have thousands of total messages, maximum results set to 1,000, but message
browser might display fewer than 1,000 results. Change **Latest** to **From beginning** to get
results from the beginning of the topic up to the 1,000 maximum result limit. You could also select a specific
time, an offset, or a timestamp to fetch a different set of results up to the maximum.
7. To browse the messages pane, use the following features:
- **Max results** drop-down list changes the maximum displayed results.
- **Auto-refresh** toggle pauses and resumes message browser updates. The topic continues to receive any messages produced during a pause, but message browser pauses the display.
- **Filter by timestamp, offset, key or value** filter results by the specified criteria. Enter a timestamp in Unix epoch format. Enter an offset, key, or any string value to filter by that value.
- Select a row to access the contents of the message. A message detail pane opens. In **Message details**, select **Key**, **Value**, or **Headers**.
- Use **From beginning**, **From offset**, **From timestamp**, or **From specific time** to refine message searches.
- By default, message browser displays the latest messages when you first
open it. To instead view the earliest messages from the topic, up to the
configured limit, select **From beginning**.
- To view messages starting from an offset, select a partition, then select
**From offset** and enter an offset.
- When you seek to a specified time, message browser reads messages from
the selected time onward until it reaches the maximum number of results.
To jump to a message timestamp, select **From timestamp** and enter the
timestamp in milliseconds.
## Advanced Message Search
Advanced Message Search lets you build a filter that refines your message
results by the fields that matter to you, such as a customer ID or a
zip code. The filter searches a topic’s full retained history, so you don’t
need to write a consumer application or a separate stream-processing
job.
#### NOTE
Advanced Message Search is available in Open Preview in Confluent Cloud.
A Preview feature is a Confluent Cloud component that is being introduced to gain
early feedback from developers. Preview features can be used for evaluation
and non-production testing purposes or to provide feedback to Confluent.
The warranty, SLA, and Support Services provisions of your agreement with
Confluent do not apply to Preview features. Confluent may discontinue
providing preview releases of the Preview features at any time in
Confluent’s’ sole discretion.
Before you use Advanced Message Search, you must register a schema for the
topic’s value in Schema Registry.
**Considerations**
- Filters are not saved. If you leave the **Messages** tab, the
Cloud Console clears your filter groups.
- Advanced Message Search runs on Confluent Cloud for Apache Flink. Confluent Cloud bills it the same
way as any other Flink statement: by Confluent Unit for Flink (CFU)
consumption for the time the search runs and any associated traffic
costs. For pricing details, see [Billing on Confluent Cloud for Apache Flink](../flink/concepts/flink-billing.md#flink-sql-billing).
- Advanced Message Search runs in your environment’s default
[compute pool](../flink/concepts/compute-pools.md#flink-sql-compute-pools). If default compute
pools are disabled, you need access to a compute pool in the same
region as the Kafka cluster.
- If the default compute pool has reached its maximum CFU limit,
select a dedicated compute pool instead.
- If the Kafka cluster uses private networking, you might need to
configure Flink for [private networking](../flink/concepts/flink-private-networking.md#flink-sql-private-networking).
- For Cloud Console access, you might need to select a
[REST endpoint](../networking/ccloud-console-access.md#ccloud-console-access-networking).
- To filter on key values, the topic must have a
[key schema](../sr/schemas-manage.md#cloud-create-key-schema).
### Build and run an advanced search
1. Navigate to your desired topic and select the **Messages** tab.
2. Click **Advanced filter**.
3. Build your first condition. Select a **field**, an **operator**, and a
**value**.
- **Field**: The message key, or a schema field. For nested fields, use
dot notation, for example `address.city`.
- **Operator**: `Equals`, `Does not equal`, `Contains`,
`Does not contain`, `Is null`, or `Is not null`.
Optionally, you can:
- Click **Add line** to add another condition to the same
filter group. Select **and** or **or** to set how it combines with the
previous condition.
- Click **Add new filter group** to add a separate group of
conditions. Select **and** or **or** to set how the group combines with
the preceding group.
- Click the edit icon next to a filter group’s name to rename it.
- Toggle **SQL Preview** to see the Flink SQL statement your filters
generate. From here you can download the statement as a
`.sql` file, copy it, or click **Edit in Flink** to continue refining
it in a Confluent Cloud for Apache Flink workspace.
4. Click **Apply**.
The generated statement replaces the contents of the **Search** field.
**Search Mode** switches to **Full**. Matching messages appear below
the search bar. Each message displays as columns by schema field,
not as raw key/value pairs.
## Logical type rendering in message browser
By default, message browser displays Apache Avro® `decimal` logical types as
raw bytes. To render decimal fields as decimal values, set the **Avro
decimal display** drop-down list in the **Topics > Messages** pane to
**Decimal**.
All other Avro logical types (such as `date`, `time-millis`,
`time-micros`, `timestamp-millis`, `timestamp-micros`,
`local-timestamp-millis`, `local-timestamp-micros`, and
`duration`) appear as their underlying Avro type. For example, a
`timestamp-millis` field appears as a `long` (milliseconds since
the Unix epoch), and a `date` field appears as an `int` (days
since the Unix epoch).
When you produce messages from Cloud Console, enter logical
type values using each type’s underlying Avro type (for example,
enter a `timestamp-millis` value as a `long`).
## Produce new messages to topics
You can produce messages to topics using Cloud Console. Use Cloud Console
for any of the following message-related tasks:
- Produce messages to a topic
- Validate messages you create for topics that enforce a schema
- Determine if a topic enforces a schema
- Add headers to messages
**Considerations**
- Cloud Console schema validation does not respect Schema Registry data contract rules.
- Cloud Console works with all Schema Registry draft versions of JSON Schema.
- JSON schemas of one draft type that reference JSON schemas of a different draft type are not fully supported
and might cause issues.
- While Protocol Buffers (protobuf) supports nested messages, Cloud Console does not. Cloud Console validates only top-level
messages. This means that, while you can define another message type inside a protobuf message,
Cloud Console only validates (and allows you to select) the top-level message.
- Proto3 version of protobuf revokes support for required fields. This limit isn’t unique to
Cloud Console. This note is a reminder that validation ignores optional fields in protobuf messages and
only applies to defined messages.
- Do not use Cloud Console to generate messages where you use bytes or fixed types as data types.
- If Cloud Console cannot produce a message for the schema, it produces an error message.
- Cloud Console requires that you wrap unions in Avro schemas to specify the type. A union type needs to specify whether the type of the content is null or string.
- Cloud Console uses the [AVSC library](https://github.com/mtth/avsc) for handling Avro schemas. This library’s
handling might differ slightly from handling by Java and librdkafka clients.
### Produce message with schema
To produce new messages with a schema:
1. From **Actions**, select **Produce new message**.
2. (Optional) Select a different schema from **Select schema** under **Key** or **Value**.
3. Enter your message or select **Generate message** to have Cloud Console create a message based on your schema.
4. (Optional) Select **Validate**.
5. (Optional) Select **Headers** and then select **Add header**.
1. Enter your header.
2. (Optional) Add more headers as necessary.
6. Select **Produce**.
### Produce message without schema
To produce new messages without a schema:
1. From **Actions**, select **Produce new message**.
2. In **Select schema** under **Key** or **Value**, select **Produce without schema** and then
select **Continue** to dismiss the prompt.
3. Enter your message.
4. (Optional) Select **Headers** and then select **Add header**.
1. Enter your header.
2. (Optional) Add more headers as necessary.
5. Select **Produce**.
### Inspect schema enforcement
To determine if the current topic enforces a schema:
1. From **Actions**, select **Produce new message**.
2. To determine if the topic enforces a schema, review **Select schema** for **Key** and **Value**:
- **No schema available** - no schema is associated with **Key** or **Value**
- **Default - -key|value** - the default schema associated with **Key** or **Value**
- ** - -key|value** - additional schemas associated with **Key** or **Value**
If multiple schemas are associated with a topic for serialization or deserialization, Cloud Console
uses a topic name strategy to organize them.
## Download topic messages
Use the bulk download option to download the currently displayed set of results.
**Considerations**
- You can download messages as JSON or CSV.
- Filter results to limit what you download.
- Use the message shown count indicator to determine how many messages of the current set you would download. For example,
if you set the maximum results to 50 (the default) and you filter 25 messages, you see a message that reads **25 of
50 messages shown**. This message indicates that you download only 25 messages.
**To download messages**:
- Select the download icon for JSON or CSV. Cloud Console downloads a file that contains the selected messages in the selected format to your downloads directory.
Example filenames:
- `Download messages as JSON_2023-12-05T00_47_57.084Z.json`
- `Download messages as CSV_2023-12-05T16_16_49.815Z.csv`
Example JSON content:
```json
[
{
"partition": 1,
"offset": 1511633,
"timestamp": 1701793270900,
"timestampType": "CREATE_TIME",
"key": 90687471,
"value": {
"viewtime": 90687471,
"userid": "User_1",
"pageid": "Page_60"
},
"headers": [
{
"key": "task.generation",
"value": "25"
},
{
"key": "task.id",
"value": "0"
},
{
"key": "current.iteration",
"value": "9068747"
}
],
"exceededFields": null
},
{
"partition": 1,
"offset": 1511610,
"timestamp": 1701793199427,
"timestampType": "CREATE_TIME",
"key": 90686071,
"value": {
"viewtime": 90686071,
"userid": "User_8",
"pageid": "Page_60"
},
"headers": [
{
"key": "task.generation",
"value": "25"
},
{
"key": "task.id",
"value": "0"
},
{
"key": "current.iteration",
"value": "9068607"
}
],
"exceededFields": null
}
]
```