4. Preparing Configuration Files
1. Kafka Client Configuration File
Create a file named kafka-client.properties with Kafka broker connection details:
```
bootstrap.servers=<kafka-bootstrap-server>:443
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="<kafka-username>" password="<kafka-password>";
ssl.protocol=TLSv1.2
ssl.truststore.location=./es-cert.p12
ssl.truststore.password=<truststore-password>
ssl.truststore.type=PKCS12
```
Where to find these values:
- bootstrap.servers: From Event Streams UI → Topics → Connect to this cluster → Bootstrap server
- username/password: From Event Streams UI → Connect to this cluster → Generate SCRAM credentials
- ssl.truststore.location: Path to your downloaded PKCS12 certificate
- ssl.truststore.password: Password provided when downloading the certificate
2. Schema Registry Configuration (Informational Only)
While you cannot use a properties file for Schema Registry SSL configuration with kafka-avro-console-producer, here's what the configuration would look like for reference:
schema.registry.url=https://<schema-registry-host>:443/apis/ccompat/v7
basic.auth.credentials.source=USER_INFO
basic.auth.user.info=<kafka-username>:<kafka-password>
schema.registry.ssl.truststore.location=./es-cert.p12
schema.registry.ssl.truststore.password=<truststore-password>
schema.registry.ssl.truststore.type=PKCS12
Where to find these values:
- schema.registry.url: From Event Streams UI → Schema Registry → Endpoint URL
- Important: Append /apis/ccompat/v7 to the base URL for Apicurio compatibility
- basic.auth.user.info: Same SCRAM credentials used for Kafka broker
Critical Configuration Requirements
Requirement 1: Pass Schema Registry SSL Properties via Command Line
The Problem:
The kafka-avro-console-producer tool has a known limitation: it does not read Schema Registry SSL properties from configuration files. This is different from Kafka broker SSL properties, which can be read from files.
The Solution:
Schema Registry SSL configuration must be passed directly as command-line arguments using --property flags:
```
--property schema.registry.ssl.truststore.location=./es-cert.p12
--property schema.registry.ssl.truststore.password=<password>
--property schema.registry.ssl.truststore.type=PKCS12
```
Why this matters:
If you omit these properties or try to include them only in a configuration file, you'll encounter SSL handshake errors like:
```
javax.net.ssl.SSLHandshakeException: PKIX path building failed
```
Requirement 2: Use Apicurio Compatibility Endpoint
The Problem:
IBM Event Streams uses Apicurio Registry as its Schema Registry implementation, not Confluent Schema Registry. Confluent clients expect a specific API structure.
The Solution:
When using Confluent clients with Apicurio Registry, the Schema Registry URL must include the compatibility endpoint:
```
/apis/ccompat/v7
```
Complete URL format:
```
https://<schema-registry-host>:443/apis/ccompat/v7
```
Why this matters:
Without the compatibility endpoint, you'll receive HTTP 404 errors:
```
Error registering Avro schema: 404 Not Found
```
The /apis/ccompat/v7 endpoint provides Confluent Schema Registry API compatibility, allowing Confluent tools to work seamlessly with Apicurio.
Complete Producer Command
Here's the full command with all required parameters:
```
kafka-avro-console-producer \
--bootstrap-server <kafka-bootstrap-server>:443 \
--topic test-avro-topic \
--producer.config kafka-client.properties \
--property schema.registry.url=https://<schema-registry-host>:443/apis/ccompat/v7 \
--property basic.auth.credentials.source=USER_INFO \
--property basic.auth.user.info=<kafka-username>:<kafka-password> \
--property schema.registry.ssl.truststore.location=./es-cert.p12 \
--property schema.registry.ssl.truststore.password=<truststore-password> \
--property schema.registry.ssl.truststore.type=PKCS12 \
--property value.schema='{"type":"record","name":"TestRecord","namespace":"com.example","fields":[{"name":"id","type":"int"},{"name":"name","type":"string"}]}'
```
Command Breakdown
| Parameter |
Purpose |
Source |
| bootstrap-server |
Kafka broker address |
Event Streams UI → Bootstrap server |
| topic |
Target Kafka topic |
Your topic name |
| producer.config |
Kafka client properties file |
Local file with broker SSL/SASL config |
| schema.registry.url |
Schema Registry endpoint |
Event Streams UI → Schema Registry + /apis/ccompat/v7 |
| basic.auth.credentials.source |
Auth method for Schema Registry |
Always USER_INFO |
| basic.auth.user.info |
Schema Registry credentials |
Same as Kafka SCRAM credentials |
| schema.registry.ssl.* |
SSL config for Schema Registry |
Must be on command line |
| value.schema |
Avro schema definition |
Your schema in JSON format |
Producing Messages
Once the producer starts successfully, you'll see a prompt:
```
>
```
Enter JSON messages that match your Avro schema. Each message should be on a single line:
```
{"id": 1, "name": "Test User"}
{"id": 2, "name": "Alice Smith"}
{"id": 3, "name": "Bob Johnson"}
```
Success indicators:
- No error messages appear
- The prompt returns immediately after each message
- Messages are accepted without validation errors
Press Ctrl+C to exit the producer.
Troubleshooting
Error: SSL Handshake Failed
Symptom:
```
javax.net.ssl.SSLHandshakeException: PKIX path building failed
```
Solution:
- Ensure Schema Registry SSL properties are passed via --property flags on the command line
- Verify the truststore file path is correct
- Confirm the truststore password is accurate
Error: 404 Not Found
Symptom:
```
Error registering Avro schema: 404 Not Found
```
Solution:
- Add /apis/ccompat/v7 to the Schema Registry URL
- Verify the Schema Registry endpoint is accessible
Error: Authentication Failed
Symptom:
```
401 Unauthorized
```
Solution:
- Verify SCRAM credentials are correct
- Ensure basic.auth.user.info format is username:password
- Check that the user has permissions to access the Schema Registry
Error: Schema Validation Failed
Symptom:
```
org.apache.kafka.common.errors.SerializationException: Error serializing Avro message
```
Solution:
- Ensure JSON messages exactly match the Avro schema structure
- Check field names and types
- Verify required fields are present
### Error: Connection Timeout
Symptom:
```
org.apache.kafka.common.errors.TimeoutException
```
Solution:
- Verify network connectivity to Kafka broker and Schema Registry
- Check firewall rules
- Confirm the bootstrap server and Schema Registry URLs are correct
Key Takeaways
- Schema Registry SSL properties cannot be read from configuration files - they must be passed as command-line --property arguments
- Apicurio Registry requires the /apis/ccompat/v7 compatibility endpoint when using Confluent clients
- Use the same SCRAM credentials for both Kafka broker and Schema Registry authentication
- JSON messages must strictly match the Avro schema - validation happens before serialization
- The kafka-client.properties file handles Kafka broker SSL/SASL configuration, but Schema Registry SSL must be on the command line
- Always verify messages with the consumer to ensure end-to-end serialization and deserialization works correctly
Example: Complete Working Setup
Here's a complete example with all files and commands:
File: kafka-client.properties
```
bootstrap.servers=my-kafka-cluster-kafka-bootstrap.event-streams.svc:443
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="kafka-user" password="kafka-password-123";
ssl.protocol=TLSv1.2
ssl.truststore.location=./es-cert.p12
ssl.truststore.password=cert-password-456
ssl.truststore.type=PKCS12
```
Command: Produce Messages
```
kafka-avro-console-producer \
--bootstrap-server my-kafka-cluster-kafka-bootstrap.event-streams.svc:443 \
--topic customer-events \
--producer.config kafka-client.properties \
--property schema.registry.url=https://my-schema-registry.event-streams.svc:443/apis/ccompat/v7 \
--property basic.auth.credentials.source=USER_INFO \
--property basic.auth.user.info=kafka-user:kafka-password-123 \
--property schema.registry.ssl.truststore.location=./es-cert.p12 \
--property schema.registry.ssl.truststore.password=cert-password-456 \
--property schema.registry.ssl.truststore.type=PKCS12 \
--property value.schema='{"type":"record","name":"CustomerEvent","namespace":"com.example.events","fields":[{"name":"customerId","type":"int"},{"name":"eventType","type":"string"},{"name":"timestamp","type":"long"}]}'
```
Input: Sample Messages
```
{"customerId": 1001, "eventType": "PURCHASE", "timestamp": 1711292957000}
{"customerId": 1002, "eventType": "LOGIN", "timestamp": 1711292958000}
{"customerId": 1003, "eventType": "LOGOUT", "timestamp": 1711292959000}
```