Secure Kafka Streams with Vault and Avro

Introduction

30 Second Summary

When a hospital monitors patient vitals in real time, a single unauthorized read or corrupted data point can trigger a false alarm or hide a deteriorating patient. Healthcare streaming systems need authentication on every connection, strict access controls per role, and enforced data contracts on every message.

In this project, you will harden an existing Kafka broker with SASL/SCRAM authentication, configure HashiCorp Vault to manage service credentials, deploy a Schema Registry enforcing Avro data contracts, and build a Python simulator that produces realistic patient monitoring data across 40 patients.

What You'll Build

A production-grade secured healthcare streaming foundation with authenticated producers, schema-validated messages, and role-based access controls.

By the end of this project, you'll have:

• A Kafka broker secured with SASL/SCRAM-SHA-256 authentication and topic-level ACLs enforced by the StandardAuthorizer.
• A Vault-managed credential system using AppRole authentication and per-service policies, with a Schema Registry enforcing Avro data contracts on 4 healthcare topic streams.
• An authenticated Python healthcare simulator generating realistic patient vitals, clinical events, and admissions across ICU, ER, and General Ward, plus a MEWS scoring consumer detecting patient deterioration in real time.
• Secret Mission: Build an ACL audit demonstration that proves an analytics-consumer cannot read raw patient vitals directly.

Are there any prerequisites?

This project assumes you have an existing Docker Compose environment running Kafka 3.7.1 in KRaft mode, along with Vault, Prometheus, Grafana, MinIO, and Iceberg containers.

You should be comfortable with basic Python, Docker Compose configuration, and Kafka producer/consumer concepts.

Harden Kafka with SASL/SCRAM-SHA-256

Your Kafka broker is currently running wide open. Anyone with network access to port 9092 can produce messages, consume data, and list topics without proving their identity.

In this step, you will lock down the broker with SASL/SCRAM-SHA-256 authentication. Every client connection must present valid credentials before the broker allows any operation.

In this step, get ready to:

• Reconfigure Kafka listeners for SASL_PLAINTEXT and enable SCRAM-SHA-256.
• Bootstrap SCRAM credentials for all service users into KRaft metadata.
• Enable the StandardAuthorizer and verify authenticated access works.

Reconfigure Kafka for SASL_PLAINTEXT listeners

Kafka uses listeners to define how clients connect to the broker. Switching from PLAINTEXT to SASL_PLAINTEXT forces every connection to authenticate before the broker processes any request.

• Switch back to VS Code with your docker-compose.yml open.
• Find your Kafka service's environment section.
• Add the following SASL-related environment variables to the Kafka service:

      # SASL/SCRAM authentication configuration
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:SASL_PLAINTEXT,PLAINTEXT:SASL_PLAINTEXT
      KAFKA_SASL_ENABLED_MECHANISMS: SCRAM-SHA-256
      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: SCRAM-SHA-256
      KAFKA_LISTENER_NAME_SASL__PLAINTEXT_SCRAM__SHA__256_SASL_JAAS_CONFIG: org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-secret";
      KAFKA_SASL_MECHANISM_CONTROLLER_PROTOCOL: SCRAM-SHA-256

What do these environment variables do?

The apache/kafka Docker image converts KAFKA_ prefixed environment variables into broker properties. Dots become underscores, and double underscores represent literal dots in property names.

• KAFKA_LISTENER_SECURITY_PROTOCOL_MAP maps both the CONTROLLER and PLAINTEXT listener names to SASL_PLAINTEXT protocol, requiring authentication on all connections.
• KAFKA_SASL_ENABLED_MECHANISMS tells the broker which SASL mechanisms to accept. We use SCRAM-SHA-256 because it hashes passwords instead of sending them in plaintext.
• KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL ensures broker-to-broker communication also uses SCRAM.
• The JAAS config line tells the broker how to authenticate itself to other brokers using the admin credentials.

Why SASL_PLAINTEXT instead of SASL_SSL?

SASL_PLAINTEXT handles authentication without TLS encryption. Since all traffic stays within the Docker network on your local machine, encryption adds complexity (keystores, truststores, certificate authorities) without security benefit.

SCRAM-SHA-256 still hashes credentials during the handshake, so passwords are never sent in cleartext. TLS can be layered on top in a later phase when the platform moves to a real network.

Bootstrap SCRAM credentials into KRaft metadata

In KRaft mode, SCRAM credentials must be written into the metadata log during storage formatting. This happens BEFORE the broker starts. You cannot add initial users after the fact without reformatting.

• You need to override the Kafka container's default startup to run kafka-storage.sh with --add-scram flags. Create a file called kafka-start.sh in your project directory by running:

touch kafka-start.sh

• Open kafka-start.sh in VS Code and paste the following script:

#!/bin/bash
# Custom entrypoint to bootstrap SCRAM credentials before Kafka starts

# Generate a cluster ID
CLUSTER_ID=$(/opt/kafka/bin/kafka-storage.sh random-uuid)

# Format storage with SCRAM credentials for all service users
/opt/kafka/bin/kafka-storage.sh format \
  --config /etc/kafka/docker/server.properties \
  --cluster-id $CLUSTER_ID \
  --add-scram 'SCRAM-SHA-256=[name=admin,password=admin-secret]' \
  --add-scram 'SCRAM-SHA-256=[name=vitals-producer,password=vitals-secret]' \
  --add-scram 'SCRAM-SHA-256=[name=clinical-producer,password=clinical-secret]' \
  --add-scram 'SCRAM-SHA-256=[name=monitoring-consumer,password=monitoring-secret]' \
  --add-scram 'SCRAM-SHA-256=[name=analytics-consumer,password=analytics-secret]'

# Start the Kafka broker
exec /opt/kafka/bin/kafka-server-start.sh /etc/kafka/docker/server.properties

What does this script do?

• It generates a fresh cluster ID using kafka-storage.sh random-uuid.
• It formats the KRaft metadata log with --add-scram flags that embed SCRAM credentials directly into the bootstrap.checkpoint file. Each flag creates one user.
• It creates five users: admin (super-user), vitals-producer, clinical-producer, monitoring-consumer, and analytics-consumer.
• Finally, it starts the broker using the standard kafka-server-start.sh command with exec so that Docker can properly manage the process lifecycle.

Why must SCRAM credentials be bootstrapped during formatting?

KRaft mode stores all cluster metadata (including SCRAM credentials) in an internal Raft log instead of ZooKeeper. The kafka-storage.sh format command writes the initial metadata snapshot. If you skip --add-scram during formatting, the broker starts with no users in its credential store, and nobody (including the broker itself) can authenticate.

After the broker is running, you can add MORE users dynamically via kafka-configs.sh --bootstrap-server using admin credentials. But the initial set must exist at boot time.

• Now update your docker-compose.yml to mount this script and use it as the Kafka container's entrypoint. Add these lines to your Kafka service definition:

    volumes:
      - ./kafka-start.sh:/opt/kafka/kafka-start.sh
    entrypoint: ["/bin/bash", "/opt/kafka/kafka-start.sh"]

What does this change?

• The volumes line mounts your local kafka-start.sh into the container at /opt/kafka/kafka-start.sh.
• The entrypoint override tells Docker to run your script instead of the image's default startup wrapper. This gives you control over the storage formatting step.

Enable the StandardAuthorizer and verify authentication

Authentication alone only proves identity. Authorization controls what each authenticated user can actually do. The StandardAuthorizer reads ACLs from KRaft metadata to enforce topic-level permissions.

• Add the authorization environment variables to the same Kafka service environment section in docker-compose.yml:

      # Authorization configuration
      KAFKA_AUTHORIZER_CLASS_NAME: org.apache.kafka.metadata.authorizer.StandardAuthorizer
      KAFKA_SUPER_USERS: User:admin

What do these settings do?

• KAFKA_AUTHORIZER_CLASS_NAME activates the StandardAuthorizer, which is the KRaft-native replacement for the old ZooKeeper-based AclAuthorizer.
• KAFKA_SUPER_USERS grants the admin user full access to all resources, bypassing ACL checks. This is necessary so you can manage topics and create ACLs for other users.

• Make the kafka-start.sh script executable by running:

chmod +x kafka-start.sh

• Stop and remove the existing Kafka container (volumes included) so it reformats with the new credentials. Run:

docker compose down -v
docker compose up -d

Why remove volumes?

KRaft metadata lives in a Docker volume. The kafka-storage.sh format command only runs on empty storage. Removing volumes with -v ensures a clean format with the new SCRAM credentials baked in.

• Wait about 15 seconds for Kafka to finish bootstrapping. Check that the container is healthy by running:

docker compose logs kafka | tail -20

• You should see log lines indicating the broker started successfully, with no authentication errors.

• Now create an admin.properties file that stores the admin SASL credentials for CLI tools. Create the file by running:

touch admin.properties

• Open admin.properties in VS Code and paste the following:

bootstrap.servers=localhost:9092
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-secret";

What is this file for?

Kafka CLI tools like kafka-topics.sh and kafka-acls.sh read client configuration from a properties file passed via --command-config. This file tells the CLI to authenticate using SCRAM-SHA-256 with the admin credentials.

• Verify that authentication is enforced by attempting to list topics WITHOUT credentials:

docker exec kafka /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092

• This command should fail or time out because no SASL credentials were provided.

• Now verify that authenticated access works by copying admin.properties into the container and using it:

docker cp admin.properties kafka:/opt/kafka/admin.properties
docker exec kafka /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092 --command-config /opt/kafka/admin.properties

• This command should succeed and return an empty list (no topics created yet) or the list of internal topics.

Kafka container not starting?

• Check docker compose logs kafka for errors. If you see a permissions error on kafka-start.sh, make sure you ran chmod +x kafka-start.sh on your host machine.
• If you see "Log directory already has formatted metadata", you need to fully remove volumes with docker compose down -v before starting again.
• If the broker crashes with SASL-related errors, double-check that the JAAS config environment variable uses double underscores for dots in the listener name (SASL__PLAINTEXT and SCRAM__SHA__256).

If you are still stuck, help me debug my Kafka SASL/SCRAM configuration in Docker Compose.

✔️ Awesome, I've got everything!

Great. Make sure you have saved both docker-compose.yml and admin.properties.

ⓧ I'd like to double check the full code

Here is the complete admin.properties file:

bootstrap.servers=localhost:9092
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-secret";

And here is the complete kafka-start.sh script:

#!/bin/bash
# Custom entrypoint to bootstrap SCRAM credentials before Kafka starts

# Generate a cluster ID
CLUSTER_ID=$(/opt/kafka/bin/kafka-storage.sh random-uuid)

# Format storage with SCRAM credentials for all service users
/opt/kafka/bin/kafka-storage.sh format \
  --config /etc/kafka/docker/server.properties \
  --cluster-id $CLUSTER_ID \
  --add-scram 'SCRAM-SHA-256=[name=admin,password=admin-secret]' \
  --add-scram 'SCRAM-SHA-256=[name=vitals-producer,password=vitals-secret]' \
  --add-scram 'SCRAM-SHA-256=[name=clinical-producer,password=clinical-secret]' \
  --add-scram 'SCRAM-SHA-256=[name=monitoring-consumer,password=monitoring-secret]' \
  --add-scram 'SCRAM-SHA-256=[name=analytics-consumer,password=analytics-secret]'

# Start the Kafka broker
exec /opt/kafka/bin/kafka-server-start.sh /etc/kafka/docker/server.properties

Your Kafka service in docker-compose.yml should now include these environment variables and the entrypoint override:

    environment:
      # ... your existing KRaft environment variables ...
      # SASL/SCRAM authentication configuration
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:SASL_PLAINTEXT,PLAINTEXT:SASL_PLAINTEXT
      KAFKA_SASL_ENABLED_MECHANISMS: SCRAM-SHA-256
      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: SCRAM-SHA-256
      KAFKA_LISTENER_NAME_SASL__PLAINTEXT_SCRAM__SHA__256_SASL_JAAS_CONFIG: org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-secret";
      KAFKA_SASL_MECHANISM_CONTROLLER_PROTOCOL: SCRAM-SHA-256
      # Authorization configuration
      KAFKA_AUTHORIZER_CLASS_NAME: org.apache.kafka.metadata.authorizer.StandardAuthorizer
      KAFKA_SUPER_USERS: User:admin
    volumes:
      - ./kafka-start.sh:/opt/kafka/kafka-start.sh
    entrypoint: ["/bin/bash", "/opt/kafka/kafka-start.sh"]

Your Kafka broker is now locked down. No service can produce or consume without presenting valid SCRAM credentials. Next up, you will configure Vault to manage these credentials securely so they never need to be hardcoded.

Configure Vault for Service Credential Management

Your Kafka broker now rejects any connection without valid SCRAM credentials. The next challenge is managing those credentials securely across multiple services.

Hardcoding passwords in config files or environment variables is fragile and risky. In this step, you will configure HashiCorp Vault as a centralized secrets manager so each service retrieves its own Kafka credentials at runtime through AppRole authentication.

In this step, get ready to:

• Enable the KV v2 secrets engine and store Kafka credentials for all service roles.
• Create AppRole authentication with per-service policies enforcing least-privilege access.
• Test the full AppRole login flow and verify cross-policy access is denied.

Enable KV v2 and store service credentials

Vault's KV v2 secrets engine stores versioned key-value pairs at configurable paths. You will store each service's Kafka username and password at a dedicated path so that policies can restrict access per service.

• Set the Vault address and authenticate with the root token by running:

# Set Vault address for CLI commands
export VAULT_ADDR='http://127.0.0.1:8200'

# Log in with the root token (use your actual root token)
vault login root

Why use the root token here?

You are acting as the Vault administrator configuring the system. In production, an operator would use a tightly scoped admin policy instead of the root token. For this local setup, the dev-mode root token simplifies initial configuration.

• Enable the KV v2 secrets engine at the secret/ path by running:

# Enable KV version 2 secrets engine at path secret/
vault secrets enable -path=secret kv-v2

• Store Kafka SCRAM credentials for each service role by running these commands:

# Store credentials for each service role
vault kv put secret/kafka/vitals-producer username="vitals-producer" password="vitals-secret"
vault kv put secret/kafka/clinical-producer username="clinical-producer" password="clinical-secret"
vault kv put secret/kafka/monitoring-consumer username="monitoring-consumer" password="monitoring-secret"
vault kv put secret/kafka/analytics-consumer username="analytics-consumer" password="analytics-secret"

What does this do?

Each vault kv put command writes a versioned secret at the specified path. The path structure secret/kafka/<role> maps directly to how your Vault policies will restrict access later.

The passwords here must match what you configured in kafka-storage.sh --add-scram in Step 1. Vault becomes the single source of truth for these credentials.

• Verify the credentials were stored correctly by reading one back:

# Read back the vitals-producer credentials to verify
vault kv get secret/kafka/vitals-producer

You should see the username and password keys displayed in the data section, along with version metadata.

Seeing 'path is already in use' error?

If secret/ is already enabled (common in Vault dev mode), skip the vault secrets enable command and proceed directly to storing credentials. Vault dev mode auto-enables KV v2 at secret/.

help me troubleshoot Vault secrets engine errors

Create AppRole auth with per-service policies

Each service should only be able to read its own credentials from Vault. You will create HCL policy files that grant read access to a single path, then bind each policy to an AppRole role.

• Create the policy files directory by running:

mkdir -p vault/policies

• Create the vitals-producer policy file by running:

cat > vault/policies/vitals-producer-policy.hcl <<EOF
path "secret/data/kafka/vitals-producer" {
  capabilities = ["read"]
}
EOF

• Create the clinical-producer policy file by running:

cat > vault/policies/clinical-producer-policy.hcl <<EOF
path "secret/data/kafka/clinical-producer" {
  capabilities = ["read"]
}
EOF

• Create the monitoring-consumer policy file by running:

cat > vault/policies/monitoring-consumer-policy.hcl <<EOF
path "secret/data/kafka/monitoring-consumer" {
  capabilities = ["read"]
}
EOF

• Create the analytics-consumer policy file by running:

cat > vault/policies/analytics-consumer-policy.hcl <<EOF
path "secret/data/kafka/analytics-consumer" {
  capabilities = ["read"]
}
EOF

• Confirm all four files were created by running:

ls vault/policies/

You should see vitals-producer-policy.hcl, clinical-producer-policy.hcl, monitoring-consumer-policy.hcl, and analytics-consumer-policy.hcl listed in the output.

Why does the policy path use secret/data/ instead of secret/?

KV v2 uses different API paths than the CLI shorthand. When you run vault kv get secret/kafka/vitals-producer, the actual API path is secret/data/kafka/vitals-producer.

Policies must reference the real API path, not the CLI convenience path. This is a common gotcha with KV v2.

• Load all four policies into Vault by running:

# Write policies into Vault
vault policy write vitals-producer-policy vault/policies/vitals-producer-policy.hcl
vault policy write clinical-producer-policy vault/policies/clinical-producer-policy.hcl
vault policy write monitoring-consumer-policy vault/policies/monitoring-consumer-policy.hcl
vault policy write analytics-consumer-policy vault/policies/analytics-consumer-policy.hcl

• Enable the AppRole auth method by running:

# Enable AppRole authentication method
vault auth enable approle

• Create AppRole roles for each service, binding the corresponding policy by running:

# Create AppRole roles with their respective policies
vault write auth/approle/role/vitals-producer-role \
    token_policies="vitals-producer-policy" \
    token_ttl=1h token_max_ttl=4h

vault write auth/approle/role/clinical-producer-role \
    token_policies="clinical-producer-policy" \
    token_ttl=1h token_max_ttl=4h

vault write auth/approle/role/monitoring-consumer-role \
    token_policies="monitoring-consumer-policy" \
    token_ttl=1h token_max_ttl=4h

vault write auth/approle/role/analytics-consumer-role \
    token_policies="analytics-consumer-policy" \
    token_ttl=1h token_max_ttl=4h

What does this do?

Each vault write auth/approle/role/<name> creates a machine-to-machine authentication role. The token_policies parameter binds a policy to any token issued through this role.

When a service authenticates as vitals-producer-role, the token it receives can only read secret/data/kafka/vitals-producer. This is least-privilege access in action.

You should see a success message for each command confirming the role was created.

Seeing 'path is already in use at approle/' error?

This means AppRole was already enabled on your Vault instance. Skip the vault auth enable approle command and proceed to creating roles.

help me troubleshoot Vault AppRole configuration

Test the AppRole login flow and verify access control

The final validation proves that AppRole authentication works end-to-end. You will generate credentials for the vitals-producer role, log in, and then attempt to read another service's secrets to confirm cross-policy access is denied.

• Retrieve the role-id and generate a secret-id for the vitals-producer role by running:

# Get the role-id for vitals-producer
vault read auth/approle/role/vitals-producer-role/role-id

# Generate a secret-id for vitals-producer
vault write -force auth/approle/role/vitals-producer-role/secret-id

What are role-id and secret-id?

The role-id is a public identifier for the AppRole (like a username). The secret-id is a private credential (like a password) that proves the client's identity.

Together they form a two-part credential for machine-to-machine authentication. Your Python simulator will use these in the next step to retrieve Kafka passwords from Vault at runtime.

• Save the role-id and secret-id values to a JSON file for later use by your Python simulator:

# Capture the credentials into shell variables
ROLE_ID=$(vault read -field=role_id auth/approle/role/vitals-producer-role/role-id)
SECRET_ID=$(vault write -force -field=secret_id auth/approle/role/vitals-producer-role/secret-id)

# Save to JSON file for the Python simulator
cat > vault/approle-credentials.json <<EOF
{
  "role_id": "$ROLE_ID",
  "secret_id": "$SECRET_ID"
}
EOF

• Confirm the file was created by running:

cat vault/approle-credentials.json

You should see a JSON object with role_id and secret_id fields populated with UUID-like values.

• Test the AppRole login by authenticating with the saved credentials:

# Test AppRole login with vitals-producer credentials
vault write auth/approle/login \
    role_id="$ROLE_ID" \
    secret_id="$SECRET_ID"

You should see a response containing a client_token, policies showing vitals-producer-policy, and a lease_duration of 1 hour.

• Verify the scoped token can read its own credentials by running:

# Store the token from the login response
APP_TOKEN=$(vault write -field=token auth/approle/login role_id="$ROLE_ID" secret_id="$SECRET_ID")

# Read vitals-producer credentials using the scoped token
VAULT_TOKEN=$APP_TOKEN vault kv get secret/kafka/vitals-producer

You should see the username and password for vitals-producer returned successfully.

• Verify cross-policy access is denied by attempting to read another service's credentials:

# Attempt to read clinical-producer credentials with vitals-producer token (should fail)
VAULT_TOKEN=$APP_TOKEN vault kv get secret/kafka/clinical-producer

You should see a permission denied error confirming that the vitals-producer token cannot access clinical-producer secrets.

Why does this matter for healthcare data?

In a healthcare platform, the principle of least privilege is critical for compliance. A compromised vitals service cannot escalate access to clinical records or patient admissions data.

Each service is cryptographically isolated to its own credential scope. This is the foundation your streaming platform builds on for every service that connects to Kafka.

Login returning 'permission denied'?

Make sure you are using the correct role_id and secret_id values from the earlier commands. Secret IDs can be single-use depending on configuration. Generate a fresh secret-id if needed by running vault write -force auth/approle/role/vitals-producer-role/secret-id again.

help me debug Vault AppRole login failures

Your Vault is now fully configured with per-service credential management and least-privilege access control. Next up, you will add Schema Registry to enforce Avro data contracts on your healthcare topics.

Add Schema Registry and Register Healthcare Schemas

Your Kafka broker is now locked down with SASL/SCRAM and Vault is managing credentials for every service role. The next piece of the governance puzzle is data contracts.

Without enforced schemas, any producer can push malformed data into a topic and silently break downstream consumers. Schema Registry solves this by requiring every message to conform to a registered Avro schema before it reaches Kafka.

In this step, get ready to:

• Add Confluent Schema Registry to your Docker Compose with SASL authentication.
• Define Avro schemas for 4 healthcare topic streams.
• Register all schemas via the REST API with BACKWARD compatibility enforcement.

Add Schema Registry to Docker Compose

Schema Registry is a Kafka client, so it needs SASL credentials to connect to your secured broker. You will add it as a new service using the admin credentials.

• In VS Code, open your docker-compose.yml file.
• Add the following schema-registry service definition below your existing Kafka service:

  schema-registry:
    image: confluentinc/cp-schema-registry:7.7.1
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      - kafka
    ports:
      - "8081:8081"
    environment:
      # Identifies this Schema Registry instance
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      # Connect to Kafka broker using SASL authentication
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092
      SCHEMA_REGISTRY_KAFKASTORE_SECURITY_PROTOCOL: SASL_PLAINTEXT
      SCHEMA_REGISTRY_KAFKASTORE_SASL_MECHANISM: SCRAM-SHA-256
      SCHEMA_REGISTRY_KAFKASTORE_SASL_JAAS_CONFIG: 'org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-secret";'

What does this configuration do?

SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS tells Schema Registry where to find your Kafka broker on the Docker network.

The SASL_PLAINTEXT security protocol, SCRAM-SHA-256 mechanism, and JAAS config match what your broker expects. Schema Registry authenticates as the admin user because it needs to create and manage the internal _schemas topic.

Why does Schema Registry need admin credentials?

Schema Registry stores all registered schemas in a compacted Kafka topic called _schemas. It needs permission to create this topic and write to it, which requires super-user access.

• Save docker-compose.yml.
• Start the Schema Registry container by running:

docker compose up -d schema-registry

• Wait about 10 seconds, then verify the service is healthy by running:

curl http://localhost:8081/config

• You should see a JSON response containing "compatibilityLevel":"BACKWARD" confirming the registry is running with BACKWARD compatibility as the default.

Schema Registry won't start?

Check the logs with docker compose logs schema-registry. If you see authentication errors, confirm the JAAS config in your compose file exactly matches the admin credentials you bootstrapped in Step 1.

Make sure the depends_on service name matches your Kafka service name in the compose file. If your Kafka service is not named kafka, update depends_on and SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS to match.

Help me debug my Schema Registry startup failure

Define Avro schemas for 4 healthcare topics

Each topic in your platform needs a formal data contract that defines exactly which fields a message must contain. You will create one .avsc file per topic.

• Create a schemas directory in your project folder by running:

mkdir schemas

• Create the vitals schema file by running:

touch schemas/vitals_raw.avsc

• Open schemas/vitals_raw.avsc in VS Code and paste the following schema:

{
  "type": "record",
  "name": "VitalsRaw",
  "namespace": "com.healthcare.streaming",
  "fields": [
    {"name": "patient_id", "type": "string"},
    {"name": "heart_rate", "type": "int"},
    {"name": "systolic_bp", "type": "int"},
    {"name": "diastolic_bp", "type": "int"},
    {"name": "spo2", "type": "float"},
    {"name": "respiratory_rate", "type": "int"},
    {"name": "temperature", "type": "float"},
    {"name": "timestamp", "type": "long"},
    {"name": "ward", "type": "string"}
  ]
}

What does this schema define?

This Avro record defines the exact shape of every patient vitals message. Each field has a strict type: heart_rate must be an integer, temperature must be a float, and timestamp must be a long (Unix epoch milliseconds).

If a producer tries to send a message missing any of these fields or using the wrong type, serialization will fail before the message ever reaches Kafka.

• Create schemas/clinical_events.avsc and paste the following:

{
  "type": "record",
  "name": "ClinicalEvent",
  "namespace": "com.healthcare.streaming",
  "fields": [
    {"name": "patient_id", "type": "string"},
    {"name": "event_type", "type": "string"},
    {"name": "description", "type": "string"},
    {"name": "practitioner_id", "type": "string"},
    {"name": "timestamp", "type": "long"},
    {"name": "ward", "type": "string"}
  ]
}

• Create schemas/patient_admissions.avsc and paste the following:

{
  "type": "record",
  "name": "PatientAdmission",
  "namespace": "com.healthcare.streaming",
  "fields": [
    {"name": "patient_id", "type": "string"},
    {"name": "event_type", "type": {"type": "enum", "name": "AdmissionEventType", "symbols": ["admit", "discharge", "transfer"]}},
    {"name": "ward", "type": "string"},
    {"name": "bed_id", "type": "string"},
    {"name": "timestamp", "type": "long"},
    {"name": "diagnosis", "type": "string"}
  ]
}

Why use an enum for event_type here?

Unlike clinical_events.avsc where event types are open-ended (medication given, lab result, procedure), admission events are strictly constrained to admit, discharge, or transfer. Using an Avro enum enforces this at the schema level, so no invalid admission type can slip through.

• Create schemas/staff_actions.avsc and paste the following:

{
  "type": "record",
  "name": "StaffAction",
  "namespace": "com.healthcare.streaming",
  "fields": [
    {"name": "staff_id", "type": "string"},
    {"name": "action_type", "type": "string"},
    {"name": "patient_id", "type": "string"},
    {"name": "timestamp", "type": "long"},
    {"name": "ward", "type": "string"}
  ]
}

• Save all four schema files.
• Confirm the files exist by running:

ls schemas/

• You should see clinical_events.avsc, patient_admissions.avsc, staff_actions.avsc, and vitals_raw.avsc listed.

Register schemas via the REST API

Schemas must be registered with the Schema Registry before any producer can use them. You will use the REST API to register each schema under its corresponding subject name.

• Register the vitals.raw-value schema by running:

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data "$(python3 -c "import json; print(json.dumps({'schema': open('schemas/vitals_raw.avsc').read()}))" )" \
  http://localhost:8081/subjects/vitals.raw-value/versions

What does this command do?

The python3 -c one-liner reads your .avsc file and wraps it in the JSON envelope that Schema Registry expects: {"schema": "<schema-as-string>"}.

The subject name vitals.raw-value follows Kafka's TopicNameStrategy: the topic name (vitals.raw) plus -value for the message value schema.

• You should see a response like {"id":1} confirming the schema was registered.
• Register the remaining three schemas by running each command:

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data "$(python3 -c "import json; print(json.dumps({'schema': open('schemas/clinical_events.avsc').read()}))" )" \
  http://localhost:8081/subjects/clinical.events-value/versions

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data "$(python3 -c "import json; print(json.dumps({'schema': open('schemas/patient_admissions.avsc').read()}))" )" \
  http://localhost:8081/subjects/patient.admissions-value/versions

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data "$(python3 -c "import json; print(json.dumps({'schema': open('schemas/staff_actions.avsc').read()}))" )" \
  http://localhost:8081/subjects/staff.actions-value/versions

• Each command should return a JSON object with an id field.
• Verify all 4 subjects are registered by running:

curl http://localhost:8081/subjects

• You should see: ["vitals.raw-value","clinical.events-value","patient.admissions-value","staff.actions-value"].

What is BACKWARD compatibility?

BACKWARD compatibility means a consumer using a newer schema version can still read data written with the previous schema version. This is the Schema Registry default.

In practice, you can add new fields with defaults, but you cannot remove existing required fields or change field types. This protects downstream consumers from being broken by upstream schema changes.

• Test backward compatibility enforcement by attempting to register an incompatible change. Run:

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "{\"type\":\"record\",\"name\":\"VitalsRaw\",\"namespace\":\"com.healthcare.streaming\",\"fields\":[{\"name\":\"patient_id\",\"type\":\"string\"},{\"name\":\"heart_rate\",\"type\":\"int\"}]}"}' \
  http://localhost:8081/subjects/vitals.raw-value/versions

• This should return an error response with error code 409 indicating the schema is incompatible, because it removes required fields from the original schema.

Registration returning an error for valid schemas?

If your valid schema registrations are failing, check that your .avsc files contain valid JSON (no trailing commas, properly quoted strings). Run python3 -c "import json; json.load(open('schemas/vitals_raw.avsc'))" to validate the JSON.

If Schema Registry returns a connection error, confirm it is still running with docker compose ps.

Help me troubleshoot Schema Registry schema registration errors

Your data contracts are now enforced. Every message that flows through your healthcare topics must conform to its registered Avro schema. Next, you will build an authenticated Python simulator that retrieves credentials from Vault and produces schema-validated patient data.

Build the Authenticated Healthcare Data Simulator

Your Kafka broker is locked down with SASL/SCRAM and your Schema Registry is enforcing data contracts on all 4 topics. Now you need applications that authenticate through Vault to produce real healthcare data.

In this step, you will build a Python simulator that retrieves its Kafka credentials from Vault (proving it is authorized), serializes messages against the registered Avro schemas (proving data conformance), and streams realistic patient telemetry across 40 simulated patients.

In this step, get ready to:

• Create a Python project with a Vault client module that authenticates via AppRole.
• Build a healthcare data simulator that produces to 4 Kafka topics with SASL authentication.
• Integrate Avro serialization so every message is validated against the Schema Registry before producing.

Create the Python project and Vault client module

Before writing the simulator itself, you need a clean dependency file and a reusable module that handles Vault authentication. The Vault client will use AppRole auth to prove its identity, then retrieve Kafka SCRAM credentials from the KV v2 secrets engine.

• Create a requirements.txt file in your project directory by running:

cat <<EOF > requirements.txt
confluent-kafka==2.14.2
hvac==2.4.0
requests
EOF

What are these dependencies?

• confluent-kafka==2.14.2 is the Python client for Kafka with built-in Avro serialization support.
• hvac==2.4.0 is the Python client for HashiCorp Vault.
• requests is a dependency used by hvac for HTTP communication with the Vault server.

• Install the dependencies by running:

pip install -r requirements.txt

• Confirm the install succeeded. You should see output ending with Successfully installed followed by the package names.

Seeing install errors?

If you see errors about missing librdkafka, the confluent-kafka package needs the C library. On macOS, try brew install librdkafka first, then re-run the pip install.

If pip itself is not found, use pip3 or python3 -m pip instead. Help me troubleshoot my pip install

• Create the vault_client.py module by pasting the following code into a new file:

import hvac


class VaultClient:
    """Client for authenticating to Vault and retrieving Kafka credentials."""

    def __init__(self, vault_addr, role_id, secret_id):
        # Initialize the hvac client with the Vault server address
        self.client = hvac.Client(url=vault_addr)
        self.role_id = role_id
        self.secret_id = secret_id

    def authenticate(self):
        """Authenticate to Vault using AppRole credentials."""
        # Login with AppRole role_id and secret_id
        response = self.client.auth.approle.login(
            role_id=self.role_id,
            secret_id=self.secret_id
        )
        # Store the client token for subsequent requests
        self.client.token = response["auth"]["client_token"]
        print(f"Authenticated to Vault. Token accessor: {response['auth']['accessor']}")

    def get_kafka_credentials(self, service_name):
        """Retrieve Kafka SCRAM credentials for a service from Vault KV v2."""
        # Read the secret from the KV v2 engine at secret/data/kafka/<service>
        secret = self.client.secrets.kv.read_secret_version(
            path=f"kafka/{service_name}"
        )
        # Extract username and password from the nested data structure
        creds = secret["data"]["data"]
        return {"username": creds["username"], "password": creds["password"]}

What does this code do?

• __init__ creates an hvac client pointed at the Vault server and stores the AppRole credentials for later use.
• authenticate() calls client.auth.approle.login() to exchange the role_id and secret_id for a Vault token.
• get_kafka_credentials() reads from the KV v2 path secret/data/kafka/<service_name> and returns a dict with username and password keys.

• Save vault_client.py.

Build the simulator with Vault authentication and patient data generation

The simulator authenticates to Vault, retrieves SCRAM credentials, configures a Kafka producer, and generates realistic patient data for 40 patients across three hospital wards. You will build it in stages, starting with imports and patient definitions.

• Create a new file called simulator.py and paste the following imports, configuration, and patient definitions:

import json
import time
import random
from confluent_kafka import Producer
from confluent_kafka.serialization import SerializationContext, MessageField
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from vault_client import VaultClient

# ============================================================
# Configuration
# ============================================================
VAULT_ADDR = "http://localhost:8200"
SCHEMA_REGISTRY_URL = "http://localhost:8081"
KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"

# Load AppRole credentials for vitals-producer
with open("vault/approle-credentials.json", "r") as f:
    approle_creds = json.load(f)

ROLE_ID = approle_creds["role_id"]
SECRET_ID = approle_creds["secret_id"]

# ============================================================
# Patient definitions: 40 patients across 3 wards
# ============================================================
PATIENTS = []

# ICU patients: P001-P010 (high acuity)
for i in range(1, 11):
    PATIENTS.append({"patient_id": f"P{i:03d}", "ward": "ICU"})

# ER patients: P011-P025 (moderate acuity)
for i in range(11, 26):
    PATIENTS.append({"patient_id": f"P{i:03d}", "ward": "ER"})

# General Ward patients: P026-P040 (lower acuity)
for i in range(26, 41):
    PATIENTS.append({"patient_id": f"P{i:03d}", "ward": "General Ward"})

What does this code do?

• The imports pull in the Kafka producer, Avro serialization classes, and the VaultClient module you just built.
• Configuration points at local Vault, Schema Registry, and Kafka. The AppRole credentials come from the JSON file you created in Step 2.
• The PATIENTS list defines 40 patients: 10 in ICU (highest acuity), 15 in ER, and 15 in General Ward. Each patient has a unique ID and ward assignment that drives vital sign ranges later.

• Below the patient definitions in the same file, add the Avro schema strings and data generation functions:

# ============================================================
# Avro schema definitions (must match registered schemas)
# ============================================================
VITALS_SCHEMA = json.dumps({
    "type": "record",
    "name": "VitalsRaw",
    "namespace": "com.healthcare.vitals",
    "fields": [
        {"name": "patient_id", "type": "string"},
        {"name": "heart_rate", "type": "int"},
        {"name": "systolic_bp", "type": "int"},
        {"name": "diastolic_bp", "type": "int"},
        {"name": "spo2", "type": "float"},
        {"name": "respiratory_rate", "type": "int"},
        {"name": "temperature", "type": "float"},
        {"name": "timestamp", "type": "long"},
        {"name": "ward", "type": "string"}
    ]
})

CLINICAL_EVENTS_SCHEMA = json.dumps({
    "type": "record",
    "name": "ClinicalEvent",
    "namespace": "com.healthcare.clinical",
    "fields": [
        {"name": "patient_id", "type": "string"},
        {"name": "event_type", "type": "string"},
        {"name": "description", "type": "string"},
        {"name": "practitioner_id", "type": "string"},
        {"name": "timestamp", "type": "long"},
        {"name": "ward", "type": "string"}
    ]
})

ADMISSIONS_SCHEMA = json.dumps({
    "type": "record",
    "name": "PatientAdmission",
    "namespace": "com.healthcare.admissions",
    "fields": [
        {"name": "patient_id", "type": "string"},
        {"name": "event_type", "type": "string"},
        {"name": "ward", "type": "string"},
        {"name": "bed_id", "type": "string"},
        {"name": "timestamp", "type": "long"},
        {"name": "diagnosis", "type": "string"}
    ]
})

STAFF_ACTIONS_SCHEMA = json.dumps({
    "type": "record",
    "name": "StaffAction",
    "namespace": "com.healthcare.staff",
    "fields": [
        {"name": "staff_id", "type": "string"},
        {"name": "action_type", "type": "string"},
        {"name": "patient_id", "type": "string"},
        {"name": "timestamp", "type": "long"},
        {"name": "ward", "type": "string"}
    ]
})

# ============================================================
# Data generation helpers
# ============================================================
def generate_vitals(patient):
    """Generate realistic vital signs based on ward acuity."""
    ward = patient["ward"]
    # ICU patients have more extreme vitals
    if ward == "ICU":
        heart_rate = random.randint(50, 140)
        systolic_bp = random.randint(70, 200)
        diastolic_bp = random.randint(40, 110)
        spo2 = round(random.uniform(88.0, 100.0), 1)
        respiratory_rate = random.randint(8, 35)
        temperature = round(random.uniform(35.0, 40.0), 1)
    elif ward == "ER":
        heart_rate = random.randint(60, 130)
        systolic_bp = random.randint(80, 180)
        diastolic_bp = random.randint(50, 100)
        spo2 = round(random.uniform(90.0, 100.0), 1)
        respiratory_rate = random.randint(10, 30)
        temperature = round(random.uniform(35.5, 39.5), 1)
    else:
        heart_rate = random.randint(60, 110)
        systolic_bp = random.randint(90, 160)
        diastolic_bp = random.randint(60, 95)
        spo2 = round(random.uniform(94.0, 100.0), 1)
        respiratory_rate = random.randint(12, 22)
        temperature = round(random.uniform(36.0, 38.5), 1)

    return {
        "patient_id": patient["patient_id"],
        "heart_rate": heart_rate,
        "systolic_bp": systolic_bp,
        "diastolic_bp": diastolic_bp,
        "spo2": spo2,
        "respiratory_rate": respiratory_rate,
        "temperature": temperature,
        "timestamp": int(time.time() * 1000),
        "ward": ward
    }

def generate_clinical_event(patient):
    """Generate a random clinical event for a patient."""
    event_types = [
        ("medication_administered", "Administered scheduled medication"),
        ("lab_ordered", "Ordered routine blood panel"),
        ("assessment", "Completed nursing assessment"),
        ("vital_alert", "Responded to vital sign alert"),
        ("procedure", "Performed bedside procedure")
    ]
    event_type, description = random.choice(event_types)
    practitioner_id = f"DR{random.randint(100, 999)}"
    return {
        "patient_id": patient["patient_id"],
        "event_type": event_type,
        "description": description,
        "practitioner_id": practitioner_id,
        "timestamp": int(time.time() * 1000),
        "ward": patient["ward"]
    }

def generate_admission_event(patient):
    """Generate an admission/discharge/transfer event."""
    event_types = ["admit", "discharge", "transfer"]
    event_type = random.choice(event_types)
    bed_id = f"{patient['ward'][:3].upper()}-{random.randint(1, 50):02d}"
    diagnoses = [
        "Acute respiratory failure",
        "Chest pain - rule out MI",
        "Sepsis",
        "Hip fracture",
        "Pneumonia",
        "Diabetic ketoacidosis"
    ]
    return {
        "patient_id": patient["patient_id"],
        "event_type": event_type,
        "ward": patient["ward"],
        "bed_id": bed_id,
        "timestamp": int(time.time() * 1000),
        "diagnosis": random.choice(diagnoses)
    }

def generate_staff_action(patient):
    """Generate a staff action event."""
    action_types = [
        "chart_review", "order_entry", "medication_scan",
        "patient_handoff", "alert_acknowledgment"
    ]
    staff_id = f"RN{random.randint(100, 999)}"
    return {
        "staff_id": staff_id,
        "action_type": random.choice(action_types),
        "patient_id": patient["patient_id"],
        "timestamp": int(time.time() * 1000),
        "ward": patient["ward"]
    }

What do the data generators do?

• The schema strings must match the Avro schemas you registered in the Schema Registry. They tell the AvroSerializer how to encode each message.
• generate_vitals() produces realistic ranges per ward. ICU patients have wider (more critical) ranges, while General Ward patients stay closer to normal.
• generate_clinical_event() and generate_admission_event() create events that simulate real hospital workflows like medication administration and patient transfers.
• generate_staff_action() simulates nursing actions that accompany clinical events.

Integrate Avro serialization and run the simulator

The final piece connects everything: Vault authentication, SASL-configured producer, Avro serializers, and the production loop that streams data to all 4 topics at realistic intervals.

• Below the data generation functions in simulator.py, add the delivery callback and main function:

# ============================================================
# Delivery callback
# ============================================================
def delivery_report(err, msg):
    """Report the result of a produce request."""
    if err is not None:
        print(f"Delivery failed: {err}")
    else:
        print(f"Delivered to {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}")

# ============================================================
# Main simulator logic
# ============================================================
def main():
    # Authenticate to Vault and retrieve Kafka credentials
    vault = VaultClient(VAULT_ADDR, ROLE_ID, SECRET_ID)
    vault.authenticate()
    creds = vault.get_kafka_credentials("vitals-producer")
    print(f"Retrieved Kafka credentials for user: {creds['username']}")

    # Configure the Kafka producer with SASL/SCRAM authentication
    producer_config = {
        "bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS,
        "security.protocol": "SASL_PLAINTEXT",
        "sasl.mechanism": "SCRAM-SHA-256",
        "sasl.username": creds["username"],
        "sasl.password": creds["password"]
    }
    producer = Producer(producer_config)

    # Configure Schema Registry client and Avro serializers
    schema_registry_conf = {"url": SCHEMA_REGISTRY_URL}
    schema_registry_client = SchemaRegistryClient(schema_registry_conf)

    vitals_serializer = AvroSerializer(
        schema_registry_client, VITALS_SCHEMA
    )
    clinical_serializer = AvroSerializer(
        schema_registry_client, CLINICAL_EVENTS_SCHEMA
    )
    admissions_serializer = AvroSerializer(
        schema_registry_client, ADMISSIONS_SCHEMA
    )
    staff_serializer = AvroSerializer(
        schema_registry_client, STAFF_ACTIONS_SCHEMA
    )

    print(f"Simulator started. Producing data for {len(PATIENTS)} patients...")
    print(f"Wards: ICU={sum(1 for p in PATIENTS if p['ward']=='ICU')}, "
          f"ER={sum(1 for p in PATIENTS if p['ward']=='ER')}, "
          f"General={sum(1 for p in PATIENTS if p['ward']=='General Ward')}")

    # Track timing for each stream type
    last_clinical_time = time.time()
    last_admission_time = time.time()
    clinical_interval = random.uniform(5, 10)
    admission_interval = random.uniform(30, 60)

    try:
        while True:
            current_time = time.time()

            # Produce vitals for all patients every 2 seconds
            for patient in PATIENTS:
                vitals = generate_vitals(patient)
                producer.produce(
                    topic="vitals.raw",
                    value=vitals_serializer(
                        vitals,
                        SerializationContext("vitals.raw", MessageField.VALUE)
                    ),
                    on_delivery=delivery_report
                )

            # Produce clinical events every 5-10 seconds
            if current_time - last_clinical_time >= clinical_interval:
                patient = random.choice(PATIENTS)
                event = generate_clinical_event(patient)
                producer.produce(
                    topic="clinical.events",
                    value=clinical_serializer(
                        event,
                        SerializationContext("clinical.events", MessageField.VALUE)
                    ),
                    on_delivery=delivery_report
                )

                # Generate a staff action alongside clinical events
                action = generate_staff_action(patient)
                producer.produce(
                    topic="staff.actions",
                    value=staff_serializer(
                        action,
                        SerializationContext("staff.actions", MessageField.VALUE)
                    ),
                    on_delivery=delivery_report
                )

                last_clinical_time = current_time
                clinical_interval = random.uniform(5, 10)

            # Produce admission events every 30-60 seconds
            if current_time - last_admission_time >= admission_interval:
                patient = random.choice(PATIENTS)
                admission = generate_admission_event(patient)
                producer.produce(
                    topic="patient.admissions",
                    value=admissions_serializer(
                        admission,
                        SerializationContext("patient.admissions", MessageField.VALUE)
                    ),
                    on_delivery=delivery_report
                )
                last_admission_time = current_time
                admission_interval = random.uniform(30, 60)

            # Flush produced messages and wait before next cycle
            producer.flush()
            time.sleep(2)

    except KeyboardInterrupt:
        print("\nSimulator stopped.")
        producer.flush()

if __name__ == "__main__":
    main()

What does the main function do?

The main() function orchestrates the full authenticated pipeline:

• It authenticates to Vault via AppRole and pulls the vitals-producer SCRAM credentials.
• It configures a Kafka Producer with SASL_PLAINTEXT and SCRAM-SHA-256 using those credentials.
• Four AvroSerializer instances handle schema validation. Each serializer is called with a SerializationContext specifying the topic and MessageField.VALUE.
• The loop produces vitals every 2 seconds (40 messages per cycle), clinical events every 5-10 seconds, and admission events every 30-60 seconds.

• Save simulator.py.
• Run the simulator by executing:

python3 simulator.py

You should see output confirming Vault authentication, credential retrieval, and delivery reports streaming to your terminal.

What should I see?

Expect output like this:

Authenticated to Vault. Token accessor: abc123...
Retrieved Kafka credentials for user: vitals-producer
Simulator started. Producing data for 40 patients...
Wards: ICU=10, ER=15, General=15
Delivered to vitals.raw [0] @ offset 0
Delivered to vitals.raw [0] @ offset 1
...

The delivery reports confirm that messages are being accepted by the broker. If schema validation fails, you would see a serialization error instead of a delivery confirmation.

• Let the simulator run for 10-15 seconds, then press Ctrl+C to stop it.
• Verify the Schema Registry shows schema usage by running:

curl http://localhost:8081/subjects

You should see all 4 subjects listed: vitals.raw-value, clinical.events-value, patient.admissions-value, and staff.actions-value.

Simulator failing to start?

• If you see a Vault connection error, confirm the Vault container is running with docker ps and check that port 8200 is exposed.
• If you see an authentication error from Vault, verify the role-id and secret-id in vault/approle-credentials.json match what you generated in Step 2.
• If you see a Kafka authentication error, confirm your Kafka container restarted with SASL enabled and that the vitals-producer SCRAM user was created during storage formatting.
• If you see a schema serialization error, check that your inline schema strings match the schemas you registered. Field names, types, and order must be identical.

Help me debug my simulator startup errors

✔️ Awesome, I've got everything!

Great. Double check you have saved both vault_client.py and simulator.py.

ⓧ I'd like to double check the full code

Here is the complete vault_client.py:

import hvac


class VaultClient:
    """Client for authenticating to Vault and retrieving Kafka credentials."""

    def __init__(self, vault_addr, role_id, secret_id):
        # Initialize the hvac client with the Vault server address
        self.client = hvac.Client(url=vault_addr)
        self.role_id = role_id
        self.secret_id = secret_id

    def authenticate(self):
        """Authenticate to Vault using AppRole credentials."""
        # Login with AppRole role_id and secret_id
        response = self.client.auth.approle.login(
            role_id=self.role_id,
            secret_id=self.secret_id
        )
        # Store the client token for subsequent requests
        self.client.token = response["auth"]["client_token"]
        print(f"Authenticated to Vault. Token accessor: {response['auth']['accessor']}")

    def get_kafka_credentials(self, service_name):
        """Retrieve Kafka SCRAM credentials for a service from Vault KV v2."""
        # Read the secret from the KV v2 engine at secret/data/kafka/<service>
        secret = self.client.secrets.kv.read_secret_version(
            path=f"kafka/{service_name}"
        )
        # Extract username and password from the nested data structure
        creds = secret["data"]["data"]
        return {"username": creds["username"], "password": creds["password"]}

Here is the complete simulator.py:

import json
import time
import random
from confluent_kafka import Producer
from confluent_kafka.serialization import SerializationContext, MessageField
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from vault_client import VaultClient

# ============================================================
# Configuration
# ============================================================
VAULT_ADDR = "http://localhost:8200"
SCHEMA_REGISTRY_URL = "http://localhost:8081"
KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"

# Load AppRole credentials for vitals-producer
with open("vault/approle-credentials.json", "r") as f:
    approle_creds = json.load(f)

ROLE_ID = approle_creds["role_id"]
SECRET_ID = approle_creds["secret_id"]

# ============================================================
# Patient definitions: 40 patients across 3 wards
# ============================================================
PATIENTS = []

# ICU patients: P001-P010 (high acuity)
for i in range(1, 11):
    PATIENTS.append({"patient_id": f"P{i:03d}", "ward": "ICU"})

# ER patients: P011-P025 (moderate acuity)
for i in range(11, 26):
    PATIENTS.append({"patient_id": f"P{i:03d}", "ward": "ER"})

# General Ward patients: P026-P040 (lower acuity)
for i in range(26, 41):
    PATIENTS.append({"patient_id": f"P{i:03d}", "ward": "General Ward"})

# ============================================================
# Avro schema definitions (must match registered schemas)
# ============================================================
VITALS_SCHEMA = json.dumps({
    "type": "record",
    "name": "VitalsRaw",
    "namespace": "com.healthcare.vitals",
    "fields": [
        {"name": "patient_id", "type": "string"},
        {"name": "heart_rate", "type": "int"},
        {"name": "systolic_bp", "type": "int"},
        {"name": "diastolic_bp", "type": "int"},
        {"name": "spo2", "type": "float"},
        {"name": "respiratory_rate", "type": "int"},
        {"name": "temperature", "type": "float"},
        {"name": "timestamp", "type": "long"},
        {"name": "ward", "type": "string"}
    ]
})

CLINICAL_EVENTS_SCHEMA = json.dumps({
    "type": "record",
    "name": "ClinicalEvent",
    "namespace": "com.healthcare.clinical",
    "fields": [
        {"name": "patient_id", "type": "string"},
        {"name": "event_type", "type": "string"},
        {"name": "description", "type": "string"},
        {"name": "practitioner_id", "type": "string"},
        {"name": "timestamp", "type": "long"},
        {"name": "ward", "type": "string"}
    ]
})

ADMISSIONS_SCHEMA = json.dumps({
    "type": "record",
    "name": "PatientAdmission",
    "namespace": "com.healthcare.admissions",
    "fields": [
        {"name": "patient_id", "type": "string"},
        {"name": "event_type", "type": "string"},
        {"name": "ward", "type": "string"},
        {"name": "bed_id", "type": "string"},
        {"name": "timestamp", "type": "long"},
        {"name": "diagnosis", "type": "string"}
    ]
})

STAFF_ACTIONS_SCHEMA = json.dumps({
    "type": "record",
    "name": "StaffAction",
    "namespace": "com.healthcare.staff",
    "fields": [
        {"name": "staff_id", "type": "string"},
        {"name": "action_type", "type": "string"},
        {"name": "patient_id", "type": "string"},
        {"name": "timestamp", "type": "long"},
        {"name": "ward", "type": "string"}
    ]
})

# ============================================================
# Data generation helpers
# ============================================================
def generate_vitals(patient):
    """Generate realistic vital signs based on ward acuity."""
    ward = patient["ward"]
    # ICU patients have more extreme vitals
    if ward == "ICU":
        heart_rate = random.randint(50, 140)
        systolic_bp = random.randint(70, 200)
        diastolic_bp = random.randint(40, 110)
        spo2 = round(random.uniform(88.0, 100.0), 1)
        respiratory_rate = random.randint(8, 35)
        temperature = round(random.uniform(35.0, 40.0), 1)
    elif ward == "ER":
        heart_rate = random.randint(60, 130)
        systolic_bp = random.randint(80, 180)
        diastolic_bp = random.randint(50, 100)
        spo2 = round(random.uniform(90.0, 100.0), 1)
        respiratory_rate = random.randint(10, 30)
        temperature = round(random.uniform(35.5, 39.5), 1)
    else:
        heart_rate = random.randint(60, 110)
        systolic_bp = random.randint(90, 160)
        diastolic_bp = random.randint(60, 95)
        spo2 = round(random.uniform(94.0, 100.0), 1)
        respiratory_rate = random.randint(12, 22)
        temperature = round(random.uniform(36.0, 38.5), 1)

    return {
        "patient_id": patient["patient_id"],
        "heart_rate": heart_rate,
        "systolic_bp": systolic_bp,
        "diastolic_bp": diastolic_bp,
        "spo2": spo2,
        "respiratory_rate": respiratory_rate,
        "temperature": temperature,
        "timestamp": int(time.time() * 1000),
        "ward": ward
    }

def generate_clinical_event(patient):
    """Generate a random clinical event for a patient."""
    event_types = [
        ("medication_administered", "Administered scheduled medication"),
        ("lab_ordered", "Ordered routine blood panel"),
        ("assessment", "Completed nursing assessment"),
        ("vital_alert", "Responded to vital sign alert"),
        ("procedure", "Performed bedside procedure")
    ]
    event_type, description = random.choice(event_types)
    practitioner_id = f"DR{random.randint(100, 999)}"
    return {
        "patient_id": patient["patient_id"],
        "event_type": event_type,
        "description": description,
        "practitioner_id": practitioner_id,
        "timestamp": int(time.time() * 1000),
        "ward": patient["ward"]
    }

def generate_admission_event(patient):
    """Generate an admission/discharge/transfer event."""
    event_types = ["admit", "discharge", "transfer"]
    event_type = random.choice(event_types)
    bed_id = f"{patient['ward'][:3].upper()}-{random.randint(1, 50):02d}"
    diagnoses = [
        "Acute respiratory failure",
        "Chest pain - rule out MI",
        "Sepsis",
        "Hip fracture",
        "Pneumonia",
        "Diabetic ketoacidosis"
    ]
    return {
        "patient_id": patient["patient_id"],
        "event_type": event_type,
        "ward": patient["ward"],
        "bed_id": bed_id,
        "timestamp": int(time.time() * 1000),
        "diagnosis": random.choice(diagnoses)
    }

def generate_staff_action(patient):
    """Generate a staff action event."""
    action_types = [
        "chart_review", "order_entry", "medication_scan",
        "patient_handoff", "alert_acknowledgment"
    ]
    staff_id = f"RN{random.randint(100, 999)}"
    return {
        "staff_id": staff_id,
        "action_type": random.choice(action_types),
        "patient_id": patient["patient_id"],
        "timestamp": int(time.time() * 1000),
        "ward": patient["ward"]
    }

# ============================================================
# Delivery callback
# ============================================================
def delivery_report(err, msg):
    """Report the result of a produce request."""
    if err is not None:
        print(f"Delivery failed: {err}")
    else:
        print(f"Delivered to {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}")

# ============================================================
# Main simulator logic
# ============================================================
def main():
    # Authenticate to Vault and retrieve Kafka credentials
    vault = VaultClient(VAULT_ADDR, ROLE_ID, SECRET_ID)
    vault.authenticate()
    creds = vault.get_kafka_credentials("vitals-producer")
    print(f"Retrieved Kafka credentials for user: {creds['username']}")

    # Configure the Kafka producer with SASL/SCRAM authentication
    producer_config = {
        "bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS,
        "security.protocol": "SASL_PLAINTEXT",
        "sasl.mechanism": "SCRAM-SHA-256",
        "sasl.username": creds["username"],
        "sasl.password": creds["password"]
    }
    producer = Producer(producer_config)

    # Configure Schema Registry client and Avro serializers
    schema_registry_conf = {"url": SCHEMA_REGISTRY_URL}
    schema_registry_client = SchemaRegistryClient(schema_registry_conf)

    vitals_serializer = AvroSerializer(
        schema_registry_client, VITALS_SCHEMA
    )
    clinical_serializer = AvroSerializer(
        schema_registry_client, CLINICAL_EVENTS_SCHEMA
    )
    admissions_serializer = AvroSerializer(
        schema_registry_client, ADMISSIONS_SCHEMA
    )
    staff_serializer = AvroSerializer(
        schema_registry_client, STAFF_ACTIONS_SCHEMA
    )

    print(f"Simulator started. Producing data for {len(PATIENTS)} patients...")
    print(f"Wards: ICU={sum(1 for p in PATIENTS if p['ward']=='ICU')}, "
          f"ER={sum(1 for p in PATIENTS if p['ward']=='ER')}, "
          f"General={sum(1 for p in PATIENTS if p['ward']=='General Ward')}")

    # Track timing for each stream type
    last_clinical_time = time.time()
    last_admission_time = time.time()
    clinical_interval = random.uniform(5, 10)
    admission_interval = random.uniform(30, 60)

    try:
        while True:
            current_time = time.time()

            # Produce vitals for all patients every 2 seconds
            for patient in PATIENTS:
                vitals = generate_vitals(patient)
                producer.produce(
                    topic="vitals.raw",
                    value=vitals_serializer(
                        vitals,
                        SerializationContext("vitals.raw", MessageField.VALUE)
                    ),
                    on_delivery=delivery_report
                )

            # Produce clinical events every 5-10 seconds
            if current_time - last_clinical_time >= clinical_interval:
                patient = random.choice(PATIENTS)
                event = generate_clinical_event(patient)
                producer.produce(
                    topic="clinical.events",
                    value=clinical_serializer(
                        event,
                        SerializationContext("clinical.events", MessageField.VALUE)
                    ),
                    on_delivery=delivery_report
                )

                # Generate a staff action alongside clinical events
                action = generate_staff_action(patient)
                producer.produce(
                    topic="staff.actions",
                    value=staff_serializer(
                        action,
                        SerializationContext("staff.actions", MessageField.VALUE)
                    ),
                    on_delivery=delivery_report
                )

                last_clinical_time = current_time
                clinical_interval = random.uniform(5, 10)

            # Produce admission events every 30-60 seconds
            if current_time - last_admission_time >= admission_interval:
                patient = random.choice(PATIENTS)
                admission = generate_admission_event(patient)
                producer.produce(
                    topic="patient.admissions",
                    value=admissions_serializer(
                        admission,
                        SerializationContext("patient.admissions", MessageField.VALUE)
                    ),
                    on_delivery=delivery_report
                )
                last_admission_time = current_time
                admission_interval = random.uniform(30, 60)

            # Flush produced messages and wait before next cycle
            producer.flush()
            time.sleep(2)

    except KeyboardInterrupt:
        print("\nSimulator stopped.")
        producer.flush()

if __name__ == "__main__":
    main()

And requirements.txt:

confluent-kafka==2.14.2
hvac==2.4.0
requests

Your simulator is streaming authenticated, schema-validated healthcare data across all 4 topics. Next, you will lock down topic access with Kafka ACLs and build a MEWS scoring consumer that detects patient deterioration in real time.

Configure ACLs and Build the MEWS Consumer

Your simulator is producing authenticated data across four topics. But right now, any authenticated user can read or write to any topic.

In a healthcare environment, that is unacceptable. A monitoring service should read vitals, but an analytics dashboard should only see derived alerts. In this step, you will lock down topic-level access with Kafka ACLs and build a MEWS consumer that detects patient deterioration in real time.

In this step, get ready to:

• Create Kafka ACLs enforcing topic-level access per service role.
• Build an authenticated MEWS scoring consumer that reads vitals and produces alerts.
• Run the full pipeline and verify real-time alert generation.

Create Kafka ACLs for topic-level access control

The StandardAuthorizer denies all operations by default unless an explicit ACL grants them. The admin super-user bypasses ACLs, but every other service must have specific permissions.

• First, create the alerts.generated topic that the MEWS consumer will write to. Run this inside the Kafka container:
• Stop the simulator first if it is running, as ACL changes can disrupt active connections.

docker exec -it kafka kafka-topics.sh --bootstrap-server localhost:9092 \
  --command-config /etc/kafka/admin.properties \
  --create --topic alerts.generated \
  --partitions 3 --replication-factor 1

What does this command do?

You are running kafka-topics.sh inside the Kafka container using admin credentials from admin.properties. The --command-config flag passes the SASL credentials to authenticate to the broker.

This creates a new topic called alerts.generated with 3 partitions. Only admin can create topics because all other users lack the required CREATE permission on the cluster.

• You should see a confirmation message that the topic was created.
• Now grant the vitals-producer write access to vitals.raw only by running:

docker exec -it kafka kafka-acls.sh --bootstrap-server localhost:9092 \
  --command-config /etc/kafka/admin.properties \
  --add --allow-principal User:vitals-producer \
  --producer --topic vitals.raw

• Grant clinical-producer write access to clinical.events and patient.admissions by running:

docker exec -it kafka kafka-acls.sh --bootstrap-server localhost:9092 \
  --command-config /etc/kafka/admin.properties \
  --add --allow-principal User:clinical-producer \
  --producer --topic clinical.events

docker exec -it kafka kafka-acls.sh --bootstrap-server localhost:9092 \
  --command-config /etc/kafka/admin.properties \
  --add --allow-principal User:clinical-producer \
  --producer --topic patient.admissions

• Grant monitoring-consumer read access to vitals.raw, clinical.events, and patient.admissions. Also grant write access to alerts.generated by running:

docker exec -it kafka kafka-acls.sh --bootstrap-server localhost:9092 \
  --command-config /etc/kafka/admin.properties \
  --add --allow-principal User:monitoring-consumer \
  --consumer --topic vitals.raw --group mews-scoring-group

docker exec -it kafka kafka-acls.sh --bootstrap-server localhost:9092 \
  --command-config /etc/kafka/admin.properties \
  --add --allow-principal User:monitoring-consumer \
  --consumer --topic clinical.events --group mews-scoring-group

docker exec -it kafka kafka-acls.sh --bootstrap-server localhost:9092 \
  --command-config /etc/kafka/admin.properties \
  --add --allow-principal User:monitoring-consumer \
  --consumer --topic patient.admissions --group mews-scoring-group

docker exec -it kafka kafka-acls.sh --bootstrap-server localhost:9092 \
  --command-config /etc/kafka/admin.properties \
  --add --allow-principal User:monitoring-consumer \
  --producer --topic alerts.generated

Why does monitoring-consumer get both consumer and producer ACLs?

The MEWS consumer reads patient vitals, scores them, and then produces alert messages to a different topic. This is a common stream-processing pattern where a service consumes from one topic and produces derived data to another.

The --consumer flag grants READ and DESCRIBE on the topic plus READ on the consumer group. The --producer flag grants WRITE, DESCRIBE, and CREATE on the target topic.

• Finally, grant analytics-consumer read-only access to alerts.generated by running:

docker exec -it kafka kafka-acls.sh --bootstrap-server localhost:9092 \
  --command-config /etc/kafka/admin.properties \
  --add --allow-principal User:analytics-consumer \
  --consumer --topic alerts.generated --group analytics-group

• Verify all ACLs are in place by listing them:

docker exec -it kafka kafka-acls.sh --bootstrap-server localhost:9092 \
  --command-config /etc/kafka/admin.properties \
  --list

• You should see a full list of ACL entries showing each principal, the operations they are allowed, and which topics they apply to.

Not seeing the ACLs listed?

Make sure admin.properties is mounted inside the container at the path you specified. Check that the file contains the correct admin SCRAM credentials. If you get an authentication error, verify the Kafka container is running with SASL enabled.

help me debug why my Kafka ACL commands are failing with authentication errors

Build the MEWS scoring consumer

The Modified Early Warning Score (MEWS) is a clinical scoring system that flags patients at risk of deterioration. Your consumer will read each vitals message, calculate a composite score, and produce an alert when the score crosses a threshold.

• Create the Avro schema for alerts. Save this as schemas/alerts_generated.avsc:

{
  "type": "record",
  "name": "MewsAlert",
  "namespace": "com.healthcare.alerts",
  "fields": [
    {"name": "patient_id", "type": "string"},
    {"name": "mews_score", "type": "int"},
    {"name": "contributing_factors", "type": {"type": "array", "items": "string"}},
    {"name": "timestamp", "type": "long"}
  ]
}

What does this schema do?

This Avro schema defines the shape of every alert message. The contributing_factors field is an array of strings listing which vitals triggered the score (e.g., high heart rate, low SpO2). This makes it easy for downstream consumers to understand why an alert fired.

• Register the alert schema with the Schema Registry by running:

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "{\"type\":\"record\",\"name\":\"MewsAlert\",\"namespace\":\"com.healthcare.alerts\",\"fields\":[{\"name\":\"patient_id\",\"type\":\"string\"},{\"name\":\"mews_score\",\"type\":\"int\"},{\"name\":\"contributing_factors\",\"type\":{\"type\":\"array\",\"items\":\"string\"}},{\"name\":\"timestamp\",\"type\":\"long\"}]}", "schemaType": "AVRO"}' \
  http://localhost:8081/subjects/alerts.generated-value/versions

• You should see a response with an id field confirming the schema was registered.
• Now create mews_consumer.py in your project directory. Start with the imports and Vault authentication:

import time
import json
from confluent_kafka import Consumer, Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer, AvroDeserializer
from confluent_kafka.serialization import SerializationContext, MessageField
from vault_client import VaultClient

# Authenticate to Vault as monitoring-consumer
vault = VaultClient(
    vault_addr="http://localhost:8200",
    role_id="MONITORING_CONSUMER_ROLE_ID",
    secret_id="MONITORING_CONSUMER_SECRET_ID"
)
vault.authenticate()
creds = vault.get_kafka_credentials("monitoring-consumer")

What does this code do?

This block imports the Kafka consumer, producer, and Avro serialization libraries. It then authenticates to Vault using the monitoring-consumer AppRole and retrieves the SCRAM credentials that will be used to connect to Kafka.

Replace MONITORING_CONSUMER_ROLE_ID and MONITORING_CONSUMER_SECRET_ID with the actual values from your Vault AppRole setup for the monitoring-consumer role.

• Next, add the Kafka consumer and producer configuration below the authentication block:

# Configure Schema Registry client
sr_client = SchemaRegistryClient({"url": "http://localhost:8081"})

# Load alert schema for serialization
with open("schemas/alerts_generated.avsc", "r") as f:
    alert_schema_str = f.read()

alert_serializer = AvroSerializer(sr_client, alert_schema_str)

# Configure authenticated Kafka consumer
consumer_conf = {
    "bootstrap.servers": "localhost:9092",
    "security.protocol": "SASL_PLAINTEXT",
    "sasl.mechanism": "SCRAM-SHA-256",
    "sasl.username": creds["username"],
    "sasl.password": creds["password"],
    "group.id": "mews-scoring-group",
    "auto.offset.reset": "latest"
}
consumer = Consumer(consumer_conf)

# Configure authenticated Kafka producer for alerts
producer_conf = {
    "bootstrap.servers": "localhost:9092",
    "security.protocol": "SASL_PLAINTEXT",
    "sasl.mechanism": "SCRAM-SHA-256",
    "sasl.username": creds["username"],
    "sasl.password": creds["password"]
}
producer = Producer(producer_conf)

What is happening here?

The consumer uses mews-scoring-group as its consumer group. This matches the ACL you created earlier that grants monitoring-consumer READ access on this group.

Both consumer and producer authenticate with the same Vault-retrieved credentials. The consumer reads vitals. The producer writes scored alerts to alerts.generated.

• Now add the MEWS scoring function that calculates the clinical risk score:

def calculate_mews(vitals):
    """Calculate Modified Early Warning Score from patient vitals."""
    score = 0
    factors = []

    # Heart rate scoring (0-3 points)
    hr = vitals["heart_rate"]
    if hr >= 130:
        score += 3
        factors.append(f"HR critically high: {hr}")
    elif hr >= 111:
        score += 2
        factors.append(f"HR high: {hr}")
    elif hr >= 101:
        score += 1
        factors.append(f"HR elevated: {hr}")

    # Systolic blood pressure scoring (0-3 points)
    sbp = vitals["systolic_bp"]
    if sbp <= 70:
        score += 3
        factors.append(f"SBP critically low: {sbp}")
    elif sbp <= 80:
        score += 2
        factors.append(f"SBP low: {sbp}")
    elif sbp <= 100:
        score += 1
        factors.append(f"SBP below normal: {sbp}")

    # Respiratory rate scoring (0-3 points)
    rr = vitals["respiratory_rate"]
    if rr >= 30:
        score += 3
        factors.append(f"RR critically high: {rr}")
    elif rr >= 21:
        score += 2
        factors.append(f"RR high: {rr}")
    elif rr >= 15:
        score += 1
        factors.append(f"RR elevated: {rr}")

    # Temperature scoring (0-2 points)
    temp = vitals["temperature"]
    if temp >= 38.5:
        score += 2
        factors.append(f"Temp high: {temp}")

    # SpO2 scoring (0-3 points)
    spo2 = vitals["spo2"]
    if spo2 < 92:
        score += 3
        factors.append(f"SpO2 critically low: {spo2}")
    elif spo2 <= 93:
        score += 2
        factors.append(f"SpO2 low: {spo2}")
    elif spo2 <= 95:
        score += 1
        factors.append(f"SpO2 below normal: {spo2}")

    return score, factors

How does MEWS scoring work?

Each vital sign gets a score from 0 to 3 based on how far it deviates from normal ranges. Heart rate of 51-100 is normal (0 points). Above 130 is critical (3 points). Systolic BP of 101-199 is normal (0 points). Below 70 is critical (3 points).

A combined score of 5 or higher triggers a clinical escalation alert. The contributing_factors list tells clinicians exactly which vitals are abnormal.

• Finally, add the main consumer loop that ties everything together:

def main():
    """Main consumer loop: read vitals, score, and produce alerts."""
    consumer.subscribe(["vitals.raw"])
    print("MEWS Consumer started. Listening for vitals...")

    # Load vitals schema for deserialization
    with open("schemas/vitals_raw.avsc", "r") as f:
        vitals_schema_str = f.read()
    vitals_deserializer = AvroDeserializer(sr_client, vitals_schema_str)

    try:
        while True:
            msg = consumer.poll(1.0)
            if msg is None:
                continue
            if msg.error():
                print(f"Consumer error: {msg.error()}")
                continue

            # Deserialize the vitals message
            vitals = vitals_deserializer(
                msg.value(),
                SerializationContext(msg.topic(), MessageField.VALUE)
            )

            # Calculate MEWS score
            mews_score, factors = calculate_mews(vitals)

            # Produce alert if score >= 5
            if mews_score >= 5:
                alert = {
                    "patient_id": vitals["patient_id"],
                    "mews_score": mews_score,
                    "contributing_factors": factors,
                    "timestamp": int(time.time() * 1000)
                }
                producer.produce(
                    topic="alerts.generated",
                    value=alert_serializer(
                        alert,
                        SerializationContext("alerts.generated", MessageField.VALUE)
                    ),
                    key=vitals["patient_id"]
                )
                producer.flush()
                print(f"ALERT: Patient {vitals['patient_id']} "
                      f"MEWS={mews_score} Factors={factors}")

    except KeyboardInterrupt:
        print("Shutting down MEWS consumer...")
    finally:
        consumer.close()


if __name__ == "__main__":
    main()

What does this loop do?

The consumer subscribes to vitals.raw and polls for new messages every second. Each message is deserialized from Avro back into a Python dictionary. The vitals are scored, and if the total MEWS is 5 or higher, an alert is produced to alerts.generated.

Using the patient_id as the message key ensures that all alerts for the same patient land on the same partition, making downstream processing simpler.

• Save mews_consumer.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved mews_consumer.py and the schemas/alerts_generated.avsc file.

ⓧ I'd like to double check the full code

import time
import json
from confluent_kafka import Consumer, Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer, AvroDeserializer
from confluent_kafka.serialization import SerializationContext, MessageField
from vault_client import VaultClient

# Authenticate to Vault as monitoring-consumer
vault = VaultClient(
    vault_addr="http://localhost:8200",
    role_id="MONITORING_CONSUMER_ROLE_ID",
    secret_id="MONITORING_CONSUMER_SECRET_ID"
)
vault.authenticate()
creds = vault.get_kafka_credentials("monitoring-consumer")

# Configure Schema Registry client
sr_client = SchemaRegistryClient({"url": "http://localhost:8081"})

# Load alert schema for serialization
with open("schemas/alerts_generated.avsc", "r") as f:
    alert_schema_str = f.read()

alert_serializer = AvroSerializer(sr_client, alert_schema_str)

# Configure authenticated Kafka consumer
consumer_conf = {
    "bootstrap.servers": "localhost:9092",
    "security.protocol": "SASL_PLAINTEXT",
    "sasl.mechanism": "SCRAM-SHA-256",
    "sasl.username": creds["username"],
    "sasl.password": creds["password"],
    "group.id": "mews-scoring-group",
    "auto.offset.reset": "latest"
}
consumer = Consumer(consumer_conf)

# Configure authenticated Kafka producer for alerts
producer_conf = {
    "bootstrap.servers": "localhost:9092",
    "security.protocol": "SASL_PLAINTEXT",
    "sasl.mechanism": "SCRAM-SHA-256",
    "sasl.username": creds["username"],
    "sasl.password": creds["password"]
}
producer = Producer(producer_conf)


def calculate_mews(vitals):
    """Calculate Modified Early Warning Score from patient vitals."""
    score = 0
    factors = []

    # Heart rate scoring (0-3 points)
    hr = vitals["heart_rate"]
    if hr >= 130:
        score += 3
        factors.append(f"HR critically high: {hr}")
    elif hr >= 111:
        score += 2
        factors.append(f"HR high: {hr}")
    elif hr >= 101:
        score += 1
        factors.append(f"HR elevated: {hr}")

    # Systolic blood pressure scoring (0-3 points)
    sbp = vitals["systolic_bp"]
    if sbp <= 70:
        score += 3
        factors.append(f"SBP critically low: {sbp}")
    elif sbp <= 80:
        score += 2
        factors.append(f"SBP low: {sbp}")
    elif sbp <= 100:
        score += 1
        factors.append(f"SBP below normal: {sbp}")

    # Respiratory rate scoring (0-3 points)
    rr = vitals["respiratory_rate"]
    if rr >= 30:
        score += 3
        factors.append(f"RR critically high: {rr}")
    elif rr >= 21:
        score += 2
        factors.append(f"RR high: {rr}")
    elif rr >= 15:
        score += 1
        factors.append(f"RR elevated: {rr}")

    # Temperature scoring (0-2 points)
    temp = vitals["temperature"]
    if temp >= 38.5:
        score += 2
        factors.append(f"Temp high: {temp}")

    # SpO2 scoring (0-3 points)
    spo2 = vitals["spo2"]
    if spo2 < 92:
        score += 3
        factors.append(f"SpO2 critically low: {spo2}")
    elif spo2 <= 93:
        score += 2
        factors.append(f"SpO2 low: {spo2}")
    elif spo2 <= 95:
        score += 1
        factors.append(f"SpO2 below normal: {spo2}")

    return score, factors


def main():
    """Main consumer loop: read vitals, score, and produce alerts."""
    consumer.subscribe(["vitals.raw"])
    print("MEWS Consumer started. Listening for vitals...")

    # Load vitals schema for deserialization
    with open("schemas/vitals_raw.avsc", "r") as f:
        vitals_schema_str = f.read()
    vitals_deserializer = AvroDeserializer(sr_client, vitals_schema_str)

    try:
        while True:
            msg = consumer.poll(1.0)
            if msg is None:
                continue
            if msg.error():
                print(f"Consumer error: {msg.error()}")
                continue

            # Deserialize the vitals message
            vitals = vitals_deserializer(
                msg.value(),
                SerializationContext(msg.topic(), MessageField.VALUE)
            )

            # Calculate MEWS score
            mews_score, factors = calculate_mews(vitals)

            # Produce alert if score >= 5
            if mews_score >= 5:
                alert = {
                    "patient_id": vitals["patient_id"],
                    "mews_score": mews_score,
                    "contributing_factors": factors,
                    "timestamp": int(time.time() * 1000)
                }
                producer.produce(
                    topic="alerts.generated",
                    value=alert_serializer(
                        alert,
                        SerializationContext("alerts.generated", MessageField.VALUE)
                    ),
                    key=vitals["patient_id"]
                )
                producer.flush()
                print(f"ALERT: Patient {vitals['patient_id']} "
                      f"MEWS={mews_score} Factors={factors}")

    except KeyboardInterrupt:
        print("Shutting down MEWS consumer...")
    finally:
        consumer.close()


if __name__ == "__main__":
    main()

Getting an import error?

If you see an error about confluent_kafka.schema_registry, make sure you installed the Avro extras. Run pip install "confluent-kafka[avro]" to install the schema registry and Avro serialization support.

help me troubleshoot import errors with confluent-kafka Avro serialization in my MEWS consumer

Run the full pipeline and verify alert generation

With ACLs in place and the MEWS consumer built, you can now run the entire pipeline end-to-end. The simulator produces vitals, the MEWS consumer scores them, and alerts appear in real time.

• Start the simulator in one terminal window by running:

python simulator.py

• Open a second terminal window and start the MEWS consumer by running:

python mews_consumer.py

• Within a few seconds, you should see ALERT messages printed by the MEWS consumer as high-acuity simulated patients trigger scores of 5 or above.
• Verify that alerts are landing in the alerts.generated topic by running this in a third terminal:

docker exec -it kafka kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --command-config /etc/kafka/admin.properties \
  --topic alerts.generated --from-beginning --max-messages 5

What should I see?

You should see binary Avro-encoded messages appearing in the console consumer output. The messages contain patient_id, mews_score, contributing_factors, and timestamp fields. ICU patients (P001-P010) are more likely to trigger alerts because the simulator generates vitals with higher variance for that ward.

• Confirm the alerts arrive within seconds of the simulator starting. The pipeline is working end-to-end.

Not seeing any alerts?

Check that the MEWS consumer shows "Listening for vitals..." without any authorization errors. If you see a GROUP_AUTHORIZATION_FAILED error, verify that the ACL for monitoring-consumer includes the correct consumer group name mews-scoring-group.

If the consumer connects but no alerts print, the simulated vitals may not be triggering the threshold. Wait 10-15 seconds for the random distributions to produce high-acuity readings.

help me debug why my MEWS consumer is not generating alerts

Your healthcare streaming platform now has production-grade access control. Every service authenticates through Vault, schemas enforce data contracts, and ACLs restrict who can read or write each topic. Next up, the Secret Mission challenges you to prove these security boundaries hold.

Secret mission

ACL Enforcement Demonstration

Your platform is secured, but how do you prove it? Build a test script that authenticates as the analytics-consumer and systematically demonstrates what it can and cannot do. Attempt to read raw patient vitals (denied), read derived alerts (allowed), and produce to any topic (denied). The result is a formatted security audit report that proves your ACL boundaries are enforced at the Kafka level.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally on Docker, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

• Docker containers (Kafka with SASL/SCRAM, Schema Registry, Vault, and supporting services).
• 5 Kafka topics with healthcare data (vitals.raw, clinical.events, patient.admissions, staff.actions, alerts.generated).
• Vault secrets, AppRole configurations, and policies (stored inside the Vault container volume).
• Local project files (Python scripts, Avro schemas, configuration files).

Keep everything running

No action needed. Choose this if you're still actively building on this platform or plan to add more consumers and producers.

• All containers continue running and consuming memory.
• The simulator and MEWS consumer continue producing and consuming messages as long as they're running in their terminal sessions.
• If you close the terminal sessions running simulator.py or mews_consumer.py, the containers still run but no new messages are produced.

Pause - I'll come back to this later

Shut down running containers to free up memory. Your data persists in Docker volumes so you can resume later.

• Stop any running Python scripts (simulator.py, mews_consumer.py) by pressing Ctrl+C in their terminal sessions.
• Stop all Docker containers by running this command from the directory containing your docker-compose.yml:

docker compose stop

• When you're ready to resume, start everything back up by running:

docker compose up -d

All Kafka topics, Vault secrets, and Schema Registry data will be intact because Docker volumes persist across stops.

Delete - I don't want to use this again

Remove all containers, volumes, and local project files for a complete cleanup.

• Stop any running Python scripts by pressing Ctrl+C in their terminal sessions.
• Remove all Docker containers and volumes by running this command from the directory containing your docker-compose.yml:

docker compose down -v

What does this command do?

The -v flag removes all named volumes declared in your compose file. This deletes all Kafka topic data, Vault secrets, Schema Registry schemas, and any other persistent state stored in Docker volumes.

• Remove the local project files you created during this project. On macOS or Linux, run:

rm -f simulator.py mews_consumer.py vault_client.py test_acl_enforcement.py requirements.txt admin.properties
rm -rf schemas/ vault/

• On Windows, run:

del simulator.py, mews_consumer.py, vault_client.py, test_acl_enforcement.py, requirements.txt, admin.properties
rmdir /s /q schemas vault

After these commands, all project resources are fully removed. Docker Desktop itself remains installed on your machine.

Nice Work!

Nice Work!

You just built a production-grade secured healthcare streaming platform from the ground up. Your Kafka broker now refuses every unauthenticated connection, Vault manages all service credentials, and Schema Registry enforces data contracts on every message.

You've learned how to:

• Harden a Kafka broker with SASL/SCRAM-SHA-256 authentication in KRaft mode, bootstrap credentials with kafka-storage.sh, and enforce topic-level access with the StandardAuthorizer and Kafka ACLs.
• Configure HashiCorp Vault with KV v2 secrets and AppRole authentication to deliver least-privilege credentials to each service role.
• Deploy Confluent Schema Registry with SASL authentication, register Avro schemas with backward compatibility, and build an authenticated Python simulator producing validated patient telemetry across 40 patients and a MEWS scoring consumer that detects patient deterioration in real time.
• Secret Mission: Demonstrated ACL enforcement by proving an analytics-consumer cannot read raw patient vitals, with a formatted security audit report.

Ready to quiz yourself?