Consume API Product

This guide describes how to validate an API product deployed in a Zilla Platform by producing and consuming messages using the Confluent Kafka command-line tools.

Prerequisites

An API Product deployed on Zilla Platform
An Application associated with the API Product
An Access Key and Secret Key generated for the Application

Download Confluent Platform Tools

Download and extract the Confluent Kafka CLI tools:

curl -O https://packages.confluent.io/archive/8.1/confluent-8.1.0.tar.gz
tar -xzf confluent-8.1.0.tar.gz

Create Client Configuration

Create a client configuration file with the required security settings:

cat <<EOF > /tmp/client.properties
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="<ACCESS_KEY>" \
  password="<SECRET_KEY>";
EOF

Produce Events

Use the producer to publish an event to the API Product topic.

CREDS="<ACCESS_KEY>:<SECRET_KEY>"

# Produce a JSON message with schema validation
echo '{"orderId":"order-123","status":"created","timestamp":1234567890000}' | \
  ./confluent-8.1.0/bin/kafka-json-schema-console-producer \
    --bootstrap-server orders.example.com:9094 \
    --topic orders.created \
    --producer.config /tmp/client.properties \
    --property schema.registry.url=https://orders.example.com:443 \
    --property basic.auth.credentials.source=USER_INFO \
    --property basic.auth.user.info="$CREDS" \
    --property auto.register.schemas=false \
    --property use.latest.version=true \
    --property value.schema='{"type":"object","properties":{"orderId":{"type":"string"},"status":{"type":"string"},"timestamp":{"type":"integer"}},"required":["orderId","status","timestamp"]}'

Note

Replace orders.example.com with your actual environment hostname and update the credentials with your API key from the console.

Consume Events

Use the consumer to read events from the same API product.

CREDS="<ACCESS_KEY>:<SECRET_KEY>"

# Consume messages with schema validation
./confluent-8.1.0/bin/kafka-json-schema-console-consumer \
  --bootstrap-server orders.example.com:9094 \
  --topic orders.created \
  --consumer.config /tmp/client.properties \
  --property schema.registry.url=https://orders.example.com:443 \
  --property basic.auth.credentials.source=USER_INFO \
  --property basic.auth.user.info="$CREDS" \
  --from-beginning

Note

Use --from-beginning to read all messages or omit it to only read new messages.