PUT /v1/topics
curl -X PUT "{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
}'
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.entity.data.topic import Topic
from metadata.generated.schema.type.schema import SchemaType, Topic as TopicSchema
metadata = OpenMetadata(
OpenMetadataConnection(
hostPort="https://your-company.open-metadata.org/api",
authProvider=AuthProvider.openmetadata,
securityConfig=OpenMetadataJWTClientConfig(jwtToken="your-jwt-token"),
)
)
# Retrieve, modify, and update
topic = metadata.get_by_id(entity=Topic, entity_id="0a021819-982e-4ee4-a5a2-4af30eaa3016")
modified = topic.model_copy(deep=True)
modified.description = "Updated: All Protobuf record events captured with enhanced schema"
modified.messageSchema = TopicSchema(
schemaType=SchemaType.Protobuf,
schemaText="syntax = \"proto3\";\npackage tutorial;\nmessage Person { ... }"
)
updated = metadata.patch(entity=Topic, source=topic, destination=modified)
print(f"Updated to version {updated.version.root}")
import org.openmetadata.schema.entity.data.Topic;
// client is an initialized OpenMetadataClient.
// Retrieve, modify, and update
Topic topic = client.topics().get("0a021819-982e-4ee4-a5a2-4af30eaa3016");
topic.setDescription("Updated: All Protobuf record events captured with enhanced schema");
Topic updated = client.topics().update(topic.getId(), topic);
# Update by ID
curl -X PATCH "{base_url}/api/v1/topics/0a021819-982e-4ee4-a5a2-4af30eaa3016" \
-H "Authorization: Bearer {access_token}" \
-H "Content-Type: application/json-patch+json" \
-d '[
{"op": "replace", "path": "/description", "value": "Updated: All Protobuf record events captured with enhanced schema"}
]'
# Update by name
curl -X PATCH "{base_url}/api/v1/topics/name/sample_kafka.address_book" \
-H "Authorization: Bearer {access_token}" \
-H "Content-Type: application/json-patch+json" \
-d '[
{"op": "replace", "path": "/description", "value": "Updated: All Protobuf record events captured with enhanced schema"}
]'
{
"id": "0a021819-982e-4ee4-a5a2-4af30eaa3016",
"name": "address_book",
"fullyQualifiedName": "sample_kafka.address_book",
"description": "Updated: All Protobuf record events captured with enhanced schema",
"version": 0.2,
"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": [],
"domains": []
}
Update a Topic
Create or update a topic with a full request body
PUT
/
v1
/
topics
PUT /v1/topics
curl -X PUT "{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
}'
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.entity.data.topic import Topic
from metadata.generated.schema.type.schema import SchemaType, Topic as TopicSchema
metadata = OpenMetadata(
OpenMetadataConnection(
hostPort="https://your-company.open-metadata.org/api",
authProvider=AuthProvider.openmetadata,
securityConfig=OpenMetadataJWTClientConfig(jwtToken="your-jwt-token"),
)
)
# Retrieve, modify, and update
topic = metadata.get_by_id(entity=Topic, entity_id="0a021819-982e-4ee4-a5a2-4af30eaa3016")
modified = topic.model_copy(deep=True)
modified.description = "Updated: All Protobuf record events captured with enhanced schema"
modified.messageSchema = TopicSchema(
schemaType=SchemaType.Protobuf,
schemaText="syntax = \"proto3\";\npackage tutorial;\nmessage Person { ... }"
)
updated = metadata.patch(entity=Topic, source=topic, destination=modified)
print(f"Updated to version {updated.version.root}")
import org.openmetadata.schema.entity.data.Topic;
// client is an initialized OpenMetadataClient.
// Retrieve, modify, and update
Topic topic = client.topics().get("0a021819-982e-4ee4-a5a2-4af30eaa3016");
topic.setDescription("Updated: All Protobuf record events captured with enhanced schema");
Topic updated = client.topics().update(topic.getId(), topic);
# Update by ID
curl -X PATCH "{base_url}/api/v1/topics/0a021819-982e-4ee4-a5a2-4af30eaa3016" \
-H "Authorization: Bearer {access_token}" \
-H "Content-Type: application/json-patch+json" \
-d '[
{"op": "replace", "path": "/description", "value": "Updated: All Protobuf record events captured with enhanced schema"}
]'
# Update by name
curl -X PATCH "{base_url}/api/v1/topics/name/sample_kafka.address_book" \
-H "Authorization: Bearer {access_token}" \
-H "Content-Type: application/json-patch+json" \
-d '[
{"op": "replace", "path": "/description", "value": "Updated: All Protobuf record events captured with enhanced schema"}
]'
{
"id": "0a021819-982e-4ee4-a5a2-4af30eaa3016",
"name": "address_book",
"fullyQualifiedName": "sample_kafka.address_book",
"description": "Updated: All Protobuf record events captured with enhanced schema",
"version": 0.2,
"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": [],
"domains": []
}
Update a Topic
Create a topic or update it if it already exists, usingPUT /v1/topics with the full request body. Send the complete definition, not just the fields you want to change. To change only some fields, use PATCH with a JSON Patch array and the application/json-patch+json content type, as shown in the PATCH examples below.
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
integer
required
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
Fully qualified names of the domains to assign for governance purposes.
array
object
Custom property values defined by your organization’s metadata schema.
PUT /v1/topics
curl -X PUT "{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
}'
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.entity.data.topic import Topic
from metadata.generated.schema.type.schema import SchemaType, Topic as TopicSchema
metadata = OpenMetadata(
OpenMetadataConnection(
hostPort="https://your-company.open-metadata.org/api",
authProvider=AuthProvider.openmetadata,
securityConfig=OpenMetadataJWTClientConfig(jwtToken="your-jwt-token"),
)
)
# Retrieve, modify, and update
topic = metadata.get_by_id(entity=Topic, entity_id="0a021819-982e-4ee4-a5a2-4af30eaa3016")
modified = topic.model_copy(deep=True)
modified.description = "Updated: All Protobuf record events captured with enhanced schema"
modified.messageSchema = TopicSchema(
schemaType=SchemaType.Protobuf,
schemaText="syntax = \"proto3\";\npackage tutorial;\nmessage Person { ... }"
)
updated = metadata.patch(entity=Topic, source=topic, destination=modified)
print(f"Updated to version {updated.version.root}")
import org.openmetadata.schema.entity.data.Topic;
// client is an initialized OpenMetadataClient.
// Retrieve, modify, and update
Topic topic = client.topics().get("0a021819-982e-4ee4-a5a2-4af30eaa3016");
topic.setDescription("Updated: All Protobuf record events captured with enhanced schema");
Topic updated = client.topics().update(topic.getId(), topic);
# Update by ID
curl -X PATCH "{base_url}/api/v1/topics/0a021819-982e-4ee4-a5a2-4af30eaa3016" \
-H "Authorization: Bearer {access_token}" \
-H "Content-Type: application/json-patch+json" \
-d '[
{"op": "replace", "path": "/description", "value": "Updated: All Protobuf record events captured with enhanced schema"}
]'
# Update by name
curl -X PATCH "{base_url}/api/v1/topics/name/sample_kafka.address_book" \
-H "Authorization: Bearer {access_token}" \
-H "Content-Type: application/json-patch+json" \
-d '[
{"op": "replace", "path": "/description", "value": "Updated: All Protobuf record events captured with enhanced schema"}
]'
{
"id": "0a021819-982e-4ee4-a5a2-4af30eaa3016",
"name": "address_book",
"fullyQualifiedName": "sample_kafka.address_book",
"description": "Updated: All Protobuf record events captured with enhanced schema",
"version": 0.2,
"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": [],
"domains": []
}
Returns
Returns the updated topic object with the new version number.Response
string
Unique identifier for the topic (UUID format).
string
Topic name.
string
Fully qualified name in format
service.topicName.string
Updated description.
number
Incremented version number.
Error Handling
| Code | Error Type | Description |
|---|---|---|
400 | BAD_REQUEST | Invalid or malformed request body |
401 | UNAUTHORIZED | Invalid or missing authentication token |
403 | FORBIDDEN | User lacks permission to update this topic |
409 | CONFLICT | Concurrent modification detected |
Was this page helpful?