POST /v1/topics

from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import (
    AuthProvider,
    OpenMetadataConnection,
)
from metadata.generated.schema.security.client.openMetadataJWTClientConfig import (
    OpenMetadataJWTClientConfig,
)
from metadata.ingestion.ometa.ometa_api import OpenMetadata
from metadata.generated.schema.api.data.createTopic import CreateTopicRequest

metadata = OpenMetadata(
    OpenMetadataConnection(
        hostPort="https://your-company.open-metadata.org/api",
        authProvider=AuthProvider.openmetadata,
        securityConfig=OpenMetadataJWTClientConfig(jwtToken="your-jwt-token"),
    )
)

request = CreateTopicRequest(
    name="address_book",
    displayName="Address Book",
    service="sample_kafka",
    description="All Protobuf record related events gets captured in this topic",
    messageSchema={
        "schemaType": "Protobuf",
        "schemaText": "syntax = \"proto2\";\npackage tutorial;\nmessage Person { ... }"
    },
    partitions=1,
    cleanupPolicies=["delete"],
    replicationFactor=1,
    maximumMessageSize=1,
    retentionSize=-1
)

topic = metadata.create_or_update(request)
print(f"Created: {topic.fullyQualifiedName.root}")

import java.util.List;
import org.openmetadata.schema.api.data.CreateTopic;
import org.openmetadata.schema.entity.data.Topic;
import org.openmetadata.schema.type.topic.CleanupPolicy;

// client is an initialized OpenMetadataClient.

CreateTopic request = new CreateTopic()
    .withName("address_book")
    .withDisplayName("Address Book")
    .withService("sample_kafka")
    .withDescription("All Protobuf record related events gets captured in this topic")
    .withPartitions(1)
    .withCleanupPolicies(List.of(CleanupPolicy.DELETE))
    .withReplicationFactor(1);

Topic topic = client.topics().create(request);

curl -X POST "{base_url}/api/v1/topics" \
  -H "Authorization: Bearer {access_token}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "address_book",
    "service": "sample_kafka",
    "description": "All Protobuf record related events gets captured in this topic",
    "messageSchema": {
      "schemaType": "Protobuf",
      "schemaText": "syntax = \"proto2\";\npackage tutorial;\nmessage Person { ... }"
    },
    "partitions": 1,
    "cleanupPolicies": ["delete"],
    "replicationFactor": 1,
    "maximumMessageSize": 1,
    "retentionSize": -1
  }'

{
  "id": "0a021819-982e-4ee4-a5a2-4af30eaa3016",
  "name": "address_book",
  "fullyQualifiedName": "sample_kafka.address_book",
  "description": "All Protobuf record related events gets captured in this topic",
  "version": 0.1,
  "updatedAt": 1769982663546,
  "updatedBy": "admin",
  "service": {
    "id": "469ef25e-9bdf-4d5f-8553-eb0ce8581f30",
    "type": "messagingService",
    "name": "sample_kafka",
    "fullyQualifiedName": "sample_kafka",
    "displayName": "sample_kafka",
    "deleted": false
  },
  "serviceType": "Kafka",
  "messageSchema": {
    "schemaText": "syntax = \"proto2\";\npackage tutorial;\nmessage Person { ... }",
    "schemaType": "Protobuf",
    "schemaFields": [
      {
        "name": "AddressBook",
        "dataType": "RECORD",
        "fullyQualifiedName": "sample_kafka.address_book.AddressBook",
        "children": []
      }
    ]
  },
  "partitions": 1,
  "cleanupPolicies": ["delete"],
  "replicationFactor": 1,
  "maximumMessageSize": 1,
  "retentionSize": -1,
  "href": "http://localhost:8585/api/v1/topics/0a021819-982e-4ee4-a5a2-4af30eaa3016",
  "deleted": false,
  "owners": [],
  "tags": [],
  "followers": [],
  "votes": {
    "upVotes": 0,
    "downVotes": 0,
    "upVoters": [],
    "downVoters": []
  },
  "domains": []
}

POST

topics

POST /v1/topics

from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import (
    AuthProvider,
    OpenMetadataConnection,
)
from metadata.generated.schema.security.client.openMetadataJWTClientConfig import (
    OpenMetadataJWTClientConfig,
)
from metadata.ingestion.ometa.ometa_api import OpenMetadata
from metadata.generated.schema.api.data.createTopic import CreateTopicRequest

metadata = OpenMetadata(
    OpenMetadataConnection(
        hostPort="https://your-company.open-metadata.org/api",
        authProvider=AuthProvider.openmetadata,
        securityConfig=OpenMetadataJWTClientConfig(jwtToken="your-jwt-token"),
    )
)

request = CreateTopicRequest(
    name="address_book",
    displayName="Address Book",
    service="sample_kafka",
    description="All Protobuf record related events gets captured in this topic",
    messageSchema={
        "schemaType": "Protobuf",
        "schemaText": "syntax = \"proto2\";\npackage tutorial;\nmessage Person { ... }"
    },
    partitions=1,
    cleanupPolicies=["delete"],
    replicationFactor=1,
    maximumMessageSize=1,
    retentionSize=-1
)

topic = metadata.create_or_update(request)
print(f"Created: {topic.fullyQualifiedName.root}")

import java.util.List;
import org.openmetadata.schema.api.data.CreateTopic;
import org.openmetadata.schema.entity.data.Topic;
import org.openmetadata.schema.type.topic.CleanupPolicy;

// client is an initialized OpenMetadataClient.

CreateTopic request = new CreateTopic()
    .withName("address_book")
    .withDisplayName("Address Book")
    .withService("sample_kafka")
    .withDescription("All Protobuf record related events gets captured in this topic")
    .withPartitions(1)
    .withCleanupPolicies(List.of(CleanupPolicy.DELETE))
    .withReplicationFactor(1);

Topic topic = client.topics().create(request);

curl -X POST "{base_url}/api/v1/topics" \
  -H "Authorization: Bearer {access_token}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "address_book",
    "service": "sample_kafka",
    "description": "All Protobuf record related events gets captured in this topic",
    "messageSchema": {
      "schemaType": "Protobuf",
      "schemaText": "syntax = \"proto2\";\npackage tutorial;\nmessage Person { ... }"
    },
    "partitions": 1,
    "cleanupPolicies": ["delete"],
    "replicationFactor": 1,
    "maximumMessageSize": 1,
    "retentionSize": -1
  }'

{
  "id": "0a021819-982e-4ee4-a5a2-4af30eaa3016",
  "name": "address_book",
  "fullyQualifiedName": "sample_kafka.address_book",
  "description": "All Protobuf record related events gets captured in this topic",
  "version": 0.1,
  "updatedAt": 1769982663546,
  "updatedBy": "admin",
  "service": {
    "id": "469ef25e-9bdf-4d5f-8553-eb0ce8581f30",
    "type": "messagingService",
    "name": "sample_kafka",
    "fullyQualifiedName": "sample_kafka",
    "displayName": "sample_kafka",
    "deleted": false
  },
  "serviceType": "Kafka",
  "messageSchema": {
    "schemaText": "syntax = \"proto2\";\npackage tutorial;\nmessage Person { ... }",
    "schemaType": "Protobuf",
    "schemaFields": [
      {
        "name": "AddressBook",
        "dataType": "RECORD",
        "fullyQualifiedName": "sample_kafka.address_book.AddressBook",
        "children": []
      }
    ]
  },
  "partitions": 1,
  "cleanupPolicies": ["delete"],
  "replicationFactor": 1,
  "maximumMessageSize": 1,
  "retentionSize": -1,
  "href": "http://localhost:8585/api/v1/topics/0a021819-982e-4ee4-a5a2-4af30eaa3016",
  "deleted": false,
  "owners": [],
  "tags": [],
  "followers": [],
  "votes": {
    "upVotes": 0,
    "downVotes": 0,
    "upVoters": [],
    "downVoters": []
  },
  "domains": []
}

Create a Topic

Create a new topic within a messaging service.

Body Parameters

string

required

Name of the topic. Must be unique within the parent messaging service.

string

required

Fully qualified name of the parent MessagingService, for example sample_kafka.

object

Schema definition for messages in the topic.

Show properties

string

Type of schema, for example Avro, Protobuf, JSON, Other.

string

The schema definition text.

integer

Number of partitions for the topic.

array

Array of cleanup policies for the topic, for example delete, compact.

integer

Replication factor for the topic.

integer

Maximum message size in bytes.

integer

Retention size in bytes. Use -1 for unlimited.

string

Human-readable display name for the topic.

string

Description of the topic in Markdown format.

array

Array of owner references (users or teams) to assign to the topic.

Show properties

string

UUID of the owner entity.

string

Type of owner entity, for example user, team.

string

Name of the owner entity.

array

Fully qualified names of the domains to assign for governance purposes.

array

Array of classification tags to apply to the topic.

Show properties

string

required

Fully qualified name of the tag.

string

Type of label, for example Manual, Derived, Propagated.

string

State of the tag, for example Suggested, Confirmed.

object

Custom property values defined by your organization’s metadata schema.

POST /v1/topics

from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import (
    AuthProvider,
    OpenMetadataConnection,
)
from metadata.generated.schema.security.client.openMetadataJWTClientConfig import (
    OpenMetadataJWTClientConfig,
)
from metadata.ingestion.ometa.ometa_api import OpenMetadata
from metadata.generated.schema.api.data.createTopic import CreateTopicRequest

metadata = OpenMetadata(
    OpenMetadataConnection(
        hostPort="https://your-company.open-metadata.org/api",
        authProvider=AuthProvider.openmetadata,
        securityConfig=OpenMetadataJWTClientConfig(jwtToken="your-jwt-token"),
    )
)

request = CreateTopicRequest(
    name="address_book",
    displayName="Address Book",
    service="sample_kafka",
    description="All Protobuf record related events gets captured in this topic",
    messageSchema={
        "schemaType": "Protobuf",
        "schemaText": "syntax = \"proto2\";\npackage tutorial;\nmessage Person { ... }"
    },
    partitions=1,
    cleanupPolicies=["delete"],
    replicationFactor=1,
    maximumMessageSize=1,
    retentionSize=-1
)

topic = metadata.create_or_update(request)
print(f"Created: {topic.fullyQualifiedName.root}")

import java.util.List;
import org.openmetadata.schema.api.data.CreateTopic;
import org.openmetadata.schema.entity.data.Topic;
import org.openmetadata.schema.type.topic.CleanupPolicy;

// client is an initialized OpenMetadataClient.

CreateTopic request = new CreateTopic()
    .withName("address_book")
    .withDisplayName("Address Book")
    .withService("sample_kafka")
    .withDescription("All Protobuf record related events gets captured in this topic")
    .withPartitions(1)
    .withCleanupPolicies(List.of(CleanupPolicy.DELETE))
    .withReplicationFactor(1);

Topic topic = client.topics().create(request);

curl -X POST "{base_url}/api/v1/topics" \
  -H "Authorization: Bearer {access_token}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "address_book",
    "service": "sample_kafka",
    "description": "All Protobuf record related events gets captured in this topic",
    "messageSchema": {
      "schemaType": "Protobuf",
      "schemaText": "syntax = \"proto2\";\npackage tutorial;\nmessage Person { ... }"
    },
    "partitions": 1,
    "cleanupPolicies": ["delete"],
    "replicationFactor": 1,
    "maximumMessageSize": 1,
    "retentionSize": -1
  }'

{
  "id": "0a021819-982e-4ee4-a5a2-4af30eaa3016",
  "name": "address_book",
  "fullyQualifiedName": "sample_kafka.address_book",
  "description": "All Protobuf record related events gets captured in this topic",
  "version": 0.1,
  "updatedAt": 1769982663546,
  "updatedBy": "admin",
  "service": {
    "id": "469ef25e-9bdf-4d5f-8553-eb0ce8581f30",
    "type": "messagingService",
    "name": "sample_kafka",
    "fullyQualifiedName": "sample_kafka",
    "displayName": "sample_kafka",
    "deleted": false
  },
  "serviceType": "Kafka",
  "messageSchema": {
    "schemaText": "syntax = \"proto2\";\npackage tutorial;\nmessage Person { ... }",
    "schemaType": "Protobuf",
    "schemaFields": [
      {
        "name": "AddressBook",
        "dataType": "RECORD",
        "fullyQualifiedName": "sample_kafka.address_book.AddressBook",
        "children": []
      }
    ]
  },
  "partitions": 1,
  "cleanupPolicies": ["delete"],
  "replicationFactor": 1,
  "maximumMessageSize": 1,
  "retentionSize": -1,
  "href": "http://localhost:8585/api/v1/topics/0a021819-982e-4ee4-a5a2-4af30eaa3016",
  "deleted": false,
  "owners": [],
  "tags": [],
  "followers": [],
  "votes": {
    "upVotes": 0,
    "downVotes": 0,
    "upVoters": [],
    "downVoters": []
  },
  "domains": []
}

Returns

Returns the created topic object with all specified properties and system-generated fields.

Response

string

Unique identifier for the topic (UUID format).

string

Topic name.

string

Fully qualified name in format service.topicName.

string

Human-readable display name.

string

Description of the topic in Markdown format.

object

Reference to the parent messaging service.

Show properties

string

UUID of the messaging service.

string

Type of entity (always messagingService).

string

Name of the messaging service.

string

Fully qualified name of the messaging service.

object

Schema definition for messages in the topic.

Show properties

string

Type of schema, for example Avro, Protobuf, JSON.

string

The schema definition text.

array

Parsed schema fields.

integer

Number of partitions for the topic.

array

Cleanup policies for the topic.

integer

Replication factor for the topic.

integer

Maximum message size in bytes.

integer

Retention size in bytes.

string

Type of messaging service, for example Kafka, Redpanda, Kinesis.

array

List of owners assigned to the topic.

Show properties

string

UUID of the owner entity.

string

Type of owner entity, for example user, team.

string

Name of the owner entity.

array

Domain assignments for governance.

array

Classification tags applied to the topic.

Show properties

string

Fully qualified name of the tag.

string

Type of label, for example Manual, Derived, Propagated.

string

State of the tag, for example Suggested, Confirmed.

object

Custom property values defined by your organization’s metadata schema.

number

Version number for the entity (starts at 0.1).

Create or Update (PUT)

Use PUT /v1/topics instead of POST to perform an upsert. If a topic with the same fullyQualifiedName already exists, it will be updated; otherwise, a new topic is created. The request body is the same as POST.

curl -X PUT "{base_url}/api/v1/topics" \
  -H "Authorization: Bearer {access_token}" \
  -H "Content-Type: application/json" \
  -d '{ ... same body as POST ... }'

PUT will not return a 409 conflict error if the entity already exists — it will update the existing entity instead.

Bulk Create or Update (PUT)

Use PUT /v1/topics/bulk to create or update multiple topics in a single request. The request body is an array of create request objects.

curl -X PUT "{base_url}/api/v1/topics/bulk" \
  -H "Authorization: Bearer {access_token}" \
  -H "Content-Type: application/json" \
  -d '[
    { "name": "topic_one", "service": "sample_kafka" },
    { "name": "topic_two", "service": "sample_kafka" }
  ]'

Error Handling

Code	Error Type	Description
`400`	`BAD_REQUEST`	Invalid request body or missing required fields
`401`	`UNAUTHORIZED`	Invalid or missing authentication token
`403`	`FORBIDDEN`	User lacks permission to create topics
`409`	`ENTITY_ALREADY_EXISTS`	Topic with same name already exists in service (POST only)

API Reference

Data Assets

Collaboration

AI SDK

SDK

Discovery

Lineage

Data Contracts

Teams And Users

Data Governance

Data Quality

Alerts

Incidents

Main Concepts

Create a Topic

Create a Topic

Body Parameters

Returns

Response

Create or Update (PUT)

Bulk Create or Update (PUT)

Error Handling