For AI agents: The documentation index is at https://docs.digitalocean.com/llms.txt. Markdown versions of pages use the same URL with index.html.md in place of the HTML page (for example, append index.html.md to the directory path instead of opening the HTML document).

Usage

client.databases.create_kafka_topic(
    database_cluster_uuid="9cc10173-e9ea-4176-9dbc-a4cee4c4ff30",
    body={
        "name": "events",
        "replication_factor": 2,
        "partition_count": 3,
        ...,
    },
)

Description

To create a topic attached to a Kafka cluster, send a POST request to /v2/databases/{database_cluster_uuid}/topics.

The result will be a JSON object with a topic key.

Parameters

database_cluster_uuid string required

A unique identifier for a database cluster.

name string required

Example: events

The name of the Kafka topic.

replication_factor integer optional

Example: 2

The number of nodes to replicate data across the cluster.

partition_count integer optional

Example: 3

The number of partitions available for the topic. On update, this value can only be increased.

config object optional

Show child properties

cleanup_policy string optional

The cleanup_policy sets the retention policy to use on log segments. 'delete' will discard old segments when retention time/size limits are reached. 'compact' will enable log compaction, resulting in retention of the latest value for each key.

One of: delete, compact, compact_delete

Default: delete

compression_type string optional

The compression_type specifies the compression type of the topic.

One of: producer, gzip, snappy, Iz4, zstd, uncompressed

Default: producer

delete_retention_ms integer optional

Example: 86400000

The delete_retention_ms specifies how long (in ms) to retain delete tombstone markers for topics.

Default: 86400000

file_delete_delay_ms integer optional

Example: 60000

The file_delete_delay_ms specifies the time (in ms) to wait before deleting a file from the filesystem.

Default: 60000

flush_messages integer optional

Example: 9223372036854776000

The flush_messages specifies the number of messages to accumulate on a log partition before messages are flushed to disk.

Default: 9223372036854776000

flush_ms integer optional

Example: 9223372036854776000

The flush_ms specifies the maximum time (in ms) that a message is kept in memory before being flushed to disk.

Default: 9223372036854776000

index_interval_bytes integer optional

Example: 4096

The index_interval_bytes specifies the number of bytes between entries being added into te offset index.

Default: 4096

max_compaction_lag_ms integer optional

Example: 9223372036854776000

The max_compaction_lag_ms specifies the maximum amount of time (in ms) that a message will remain uncompacted. This is only applicable if the logs are have compaction enabled.

Default: 9223372036854776000

max_message_bytes integer optional

Example: 1048588

The max_messages_bytes specifies the largest record batch size (in bytes) that can be sent to the server. This is calculated after compression if compression is enabled.

Default: 1048588

message_down_conversion_enable boolean optional

Example: True

The message_down_conversion_enable specifies whether down-conversion of message formats is enabled to satisfy consumer requests. When 'false', the broker will not perform conversion for consumers expecting older message formats. The broker will respond with an UNSUPPORTED_VERSION error for consume requests from these older clients.

Default: True

message_format_version string optional

The message_format_version specifies the message format version used by the broker to append messages to the logs. The value of this setting is assumed to be 3.0-IV1 if the broker protocol version is 3.0 or higher. By setting a particular message format version, all existing messages on disk must be smaller or equal to the specified version.

One of: 0.8.0, 0.8.1, 0.8.2, 0.9.0, 0.10.0-IV0, 0.10.0-IV1, 0.10.1-IV0, 0.10.1-IV1, 0.10.1-IV2, 0.10.2-IV0, 0.11.0-IV0, 0.11.0-IV1, 0.11.0-IV2, 1.0-IV0, 1.1-IV0, 2.0-IV0, 2.0-IV1, 2.1-IV0, 2.1-IV1, 2.1-IV2, 2.2-IV0, 2.2-IV1, 2.3-IV0, 2.3-IV1, 2.4-IV0, 2.4-IV1, 2.5-IV0, 2.6-IV0, 2.7-IV0, 2.7-IV1, 2.7-IV2, 2.8-IV0, 2.8-IV1, 3.0-IV0, 3.0-IV1, 3.1-IV0, 3.2-IV0, 3.3-IV0, 3.3-IV1, 3.3-IV2, 3.3-IV3

Default: 3.0-IV1

message_timestamp_type string optional

The message_timestamp_type specifies whether to use the message create time or log append time as the timestamp on a message.

One of: create_time, log_append_time

Default: create_time

min_cleanable_dirty_ratio number optional

Example: 0.5

The min_cleanable_dirty_ratio specifies the frequency of log compaction (if enabled) in relation to duplicates present in the logs. For example, at 0.5, at most 50% of the log could be duplicates before compaction would begin.

Min: 0

Max: 1

Default: 0.5

min_compaction_lag_ms integer optional

Example: 0

The min_compaction_lag_ms specifies the minimum time (in ms) that a message will remain uncompacted in the log. Only relevant if log compaction is enabled.

Default: 0

min_insync_replicas integer optional

Example: 1

The min_insync_replicas specifies the number of replicas that must ACK a write for the write to be considered successful.

Min: 1

Default: 1

preallocate boolean optional

Example: False

The preallocate specifies whether a file should be preallocated on disk when creating a new log segment.

Default: False

retention_bytes integer optional

Example: 1000000

The retention_bytes specifies the maximum size of the log (in bytes) before deleting messages. -1 indicates that there is no limit.

Default: -1

retention_ms integer optional

Example: 604800000

The retention_ms specifies the maximum amount of time (in ms) to keep a message before deleting it.

Default: 604800000

segment_bytes integer optional

Example: 209715200

The segment_bytes specifies the maximum size of a single log file (in bytes).

Min: 14

Default: 209715200

segment_jitter_ms integer optional

Example: 0

The segment_jitter_ms specifies the maximum random jitter subtracted from the scheduled segment roll time to avoid thundering herds of segment rolling.

Default: 0

segment_ms integer optional

Example: 604800000

The segment_ms specifies the period of time after which the log will be forced to roll if the segment file isn't full. This ensures that retention can delete or compact old data.

Min: 1

Default: 604800000

Request Sample

import os
from pydo import Client

client = Client(token=os.environ.get("DIGITALOCEAN_TOKEN"))

req = {
  "name": "customer-events",
  "partitions": 3,
  "replication": 2,
  "config": {
    "retention_bytes": -1,
    "retention_ms": 100000
  }
}

resp = client.databases.create_kafka_topic(database_cluster_uuid="9cc10173-e9ea-4176-9dbc-a4cee4c4ff30", body=req)

Response Example

{
  "topic": {
    "name": "customer-events",
    "partitions": [
      {
        "size": 4096,
        "id": 0,
        "in_sync_replicas": 3,
        "earliest_offset": 0,
        "consumer_groups": [
          {
            "name": "consumer-group-1",
            "offset": 0
          },
          {
            "name": "consumer-group-2",
            "offset": 1
          }
        ]
      },
      {
        "size": 4096,
        "id": 1,
        "in_sync_replicas": 3,
        "earliest_offset": 0,
        "consumer_groups": null
      }
    ],
    "replication_factor": 3,
    "state": "active",
    "config": {
      "cleanup_policy": "delete",
      "compression_type": "producer",
      "delete_retention_ms": 86400000,
      "file_delete_delay_ms": 60000,
      "flush_messages": 9223372036854776000,
      "flush_ms": 9223372036854776000,
      "index_interval_bytes": 4096,
      "max_compaction_lag_ms": 9223372036854776000,
      "max_message_bytes": 1048588,
      "message_down_conversion_enable": true,
      "message_format_version": "3.0-IV1",
      "message_timestamp_difference_max_ms": 9223372036854776000,
      "message_timestamp_type": "create_time",
      "min_cleanable_dirty_ratio": 0.5,
      "min_compaction_lag_ms": 0,
      "min_insync_replicas": 1,
      "preallocate": false,
      "retention_bytes": -1,
      "retention_ms": 604800000,
      "segment_bytes": 209715200,
      "segment_index_bytes": 10485760,
      "segment_jitter_ms": 0,
      "segment_ms": 604800000
    }
  }
}

More Information

See /v2/databases/{database_cluster_uuid}/topics in the API reference for additional detail on responses, headers, parameters, and more.