records

Creates, updates, deletes, gets or lists a records resource.

Overview

Name	records
Type	Resource
Id	confluent.kafka.records

Fields

The following fields are returned by SELECT queries:

SELECT not supported for this resource, use SHOW METHODS to view available operations for the resource.

Methods

The following methods are available for this resource:

Name	Accessible by	Required Params	Optional Params	Description
produce_record	insert	cluster_id, topic_name, kafka_endpoint_id, region, cloud_provider		Produce records to the given topic, returning delivery reports for each record produced. This API can be used in streaming mode by setting "Transfer-Encoding: chunked" header. For as long as the connection is kept open, the server will keep accepting records. Records are streamed to and from the server as Concatenated JSON. For each record sent to the server, the server will asynchronously send back a delivery report, in the same order, each with its own error_code. An error_code of 200 indicates success. The HTTP status code will be HTTP 200 OK as long as the connection is successfully established. To identify records that have encountered an error, check the error_code of each delivery report. Note that the cluster_id is validated only when running in Confluent Cloud. This API currently does not support Schema Registry integration. Sending schemas is not supported. Only BINARY, JSON, and STRING formats are supported.

Parameters

Parameters can be passed in the WHERE clause of a query. Check the Methods section to see which parameters are required or optional for each operation.

Name	Datatype	Description
cloud_provider	string	Cloud provider, lowercase: aws, gcp, or azure (from the cluster spec.cloud). (default: cloud)
cluster_id	string	The Kafka cluster ID. (example: cluster-1)
kafka_endpoint_id	string	Per-cluster Kafka REST endpoint ID (the pkc-* host prefix from the Confluent UI Cluster -> Overview -> REST endpoint, or extract from confluent.managed_kafka_clusters.clusters spec.http_endpoint). (default: pkc-00000)
region	string	Cloud region the cluster runs in, e.g. ap-southeast-2 (from the cluster spec.region). (default: region)
topic_name	string	The topic name. (example: topic-1)

INSERT examples

produce_record

Produce records to the given topic, returning delivery reports for each
record produced. This API can be used in streaming mode by setting
"Transfer-Encoding: chunked" header. For as long as the connection is
kept open, the server will keep accepting records. Records are streamed
to and from the server as Concatenated JSON. For each record sent to the
server, the server will asynchronously send back a delivery report, in
the same order, each with its own error_code. An error_code of 200
indicates success. The HTTP status code will be HTTP 200 OK as long as
the connection is successfully established. To identify records that
have encountered an error, check the error_code of each delivery report.

Note that the cluster_id is validated only when running in Confluent Cloud.

This API currently does not support Schema Registry integration. Sending
schemas is not supported. Only BINARY, JSON, and STRING formats are
supported.

INSERT INTO confluent.kafka.records (
partition_id,
headers,
key,
value,
timestamp,
cluster_id,
topic_name,
kafka_endpoint_id,
region,
cloud_provider
)
SELECT 
{{ partition_id }},
'{{ headers }}',
'{{ key }}',
'{{ value }}',
'{{ timestamp }}',
'{{ cluster_id }}',
'{{ topic_name }}',
'{{ kafka_endpoint_id }}',
'{{ region }}',
'{{ cloud_provider }}'
RETURNING
cluster_id,
partition_id,
topic_name,
error_code,
key,
message,
offset,
timestamp,
value
;

Manifest

# Description fields are for documentation purposes
- name: records
props:
  - name: cluster_id
    value: "{{ cluster_id }}"
    description: Required parameter for the records resource.
  - name: topic_name
    value: "{{ topic_name }}"
    description: Required parameter for the records resource.
  - name: kafka_endpoint_id
    value: "{{ kafka_endpoint_id }}"
    description: Required parameter for the records resource.
  - name: region
    value: "{{ region }}"
    description: Required parameter for the records resource.
  - name: cloud_provider
    value: "{{ cloud_provider }}"
    description: Required parameter for the records resource.
  - name: partition_id
    value: {{ partition_id }}
  - name: headers
    value:
      - name: "{{ name }}"
        value: "{{ value }}"
  - name: key
    value:
      type: "{{ type }}"
      data: "{{ data }}"
  - name: value
    value:
      type: "{{ type }}"
      data: "{{ data }}"
  - name: timestamp
    value: "{{ timestamp }}"