Local Development: Redpanda Kafka + Tinybird with Docker Compose

# Install Tinybird CLI and log in
curl https://tinybird.co | sh
tb login
# Start Tinybird Local
tb local start
# Follow along to connect a Kafka topic to a Tinybird data source
tb connection create kafka
# Create analytical endpoints
tb create --prompt "an endpoint that returns latest value and today's average of the selected sensor"
# Deploy to Tinybird Cloud
tb --cloud deploy

networks:
  redpanda_network:
    driver: bridge

volumes:
  redpanda-0:

services:
  redpanda:
    image: docker.redpanda.com/redpandadata/redpanda:v25.1.4
    container_name: redpanda
    platform: linux/amd64
    command:
      - redpanda
      - start
      - --kafka-addr=internal://0.0.0.0:9092,external://0.0.0.0:19092
      - --advertise-kafka-addr=internal://redpanda:9092,external://localhost:19092
      - --pandaproxy-addr=internal://0.0.0.0:8082,external://0.0.0.0:18082
      - --advertise-pandaproxy-addr=internal://redpanda:8082,external://localhost:18082
      - --schema-registry-addr=internal://0.0.0.0:8081,external://0.0.0.0:18081
      - --rpc-addr=redpanda:33145
      - --advertise-rpc-addr=redpanda:33145
      - --mode=dev-container
      - --smp=1
      - --default-log-level=info
    environment:
      RP_BOOTSTRAP_USER: "superuser:secretpassword"
    ports:
      - "18081:18081"
      - "18082:18082"
      - "19092:19092"
      - "19644:9644"
    volumes:
      - redpanda-0:/var/lib/redpanda/data
    networks:
      - redpanda_network
    healthcheck:
      test: ["CMD", "rpk", "cluster", "info", "-X", "user=superuser", "-X", "pass=secretpassword"]
      interval: 10s
      timeout: 15s
      retries: 10

  tinybird-local:
    image: tinybirdco/tinybird-local:latest
    container_name: tinybird-local-rp
    platform: linux/amd64
    ports:
      - "7181:7181"
    networks:
      - redpanda_network
    depends_on:
      - redpanda

TYPE kafka
KAFKA_BOOTSTRAP_SERVERS {{ tb_secret("KAFKA_PROD_SERVER", "redpanda:9092") }}
KAFKA_SECURITY_PROTOCOL {{ tb_secret("KAFKA_PROD_SECURITY_PROTOCOL", "PLAINTEXT") }}
KAFKA_SASL_MECHANISM {{ tb_secret("KAFKA_PROD_SASL_MECHANISM", "PLAIN") }}
KAFKA_KEY {{ tb_secret("KAFKA_KEY", "superuser") }}
KAFKA_SECRET {{ tb_secret("KAFKA_SECRET", "secretpassword") }}

#!/bin/bash

# Function to check if Redpanda is ready
check_redpanda_health() {
    local health_output
    health_output=$(docker exec -i redpanda rpk cluster health --format json)
    if [[ $health_output == *"\"is_healthy\":true"* ]]; then
        return 0
    else
        return 1
    fi
}

# Function to check if Tinybird is ready
check_tinybird_status() {
    local status_output
    status_output=$(tb local status)
    if [[ $status_output == *"✓ Tinybird Local is ready!"* ]]; then
        return 0
    else
        return 1
    fi
}

# Maximum number of retries
MAX_RETRIES=30
RETRY_INTERVAL=5

# Wait for both Redpanda and Tinybird to be ready
echo "Waiting for both Redpanda and Tinybird Local to be ready..."

# Try until success or max retries reached
for ((i=1; i<=MAX_RETRIES; i++)); do
    if check_redpanda_health && check_tinybird_status; then
        echo "✓ Both Redpanda and Tinybird Local are ready!"
        echo "Proceeding with deployment..."
        # Create topic in Redpanda
        docker exec -i redpanda rpk topic create water_metrics_demo -X brokers=redpanda:9092
        # Deploy to Tinybird
        tb deploy
        # Produce to Redpanda topic
        cat fixtures/kafka_water_meters.ndjson | docker exec -i redpanda rpk topic produce water_metrics_demo -X brokers=redpanda:9092
        # Execute SQL query in Tinybird
        tb sql "select meter_id, timestamp, flow_rate, temperature from kafka_water_meters"
        echo -e "\n✅ Setup completed successfully!"
        exit 0
    else
        echo "Attempt $i/$MAX_RETRIES: Services not ready yet. Waiting ${RETRY_INTERVAL}s..."
        sleep $RETRY_INTERVAL
    fi
done

echo "❌ Failed to connect to services after $MAX_RETRIES attempts"
exit 1

chmod +x setup-local.sh
docker-compose up -d
./setup-local.sh

tb token copy "admin local_testing@tinybird.co" && TB_LOCAL_TOKEN=$(pbpaste)

curl -H "Authorization: Bearer $TB_LOCAL_TOKEN" -X GET "http://localhost:7181/v0/pipes/meter_measurements.json"

import os
import subprocess
import time
import requests
import yaml
import pytest
import json
import glob

# Get the tb command from environment variable or use default
TB_COMMAND = os.getenv("TB_COMMAND", os.path.expanduser("~/.local/bin/tb"))

def check_local_server():
    """Check if the local Tinybird server is running."""
    try:
        response = requests.get("http://localhost:7181/tokens")
        return response.status_code == 200
    except requests.exceptions.ConnectionError:
        return False

def load_test_cases(yaml_file):
    """Load test cases from a YAML file."""
    with open(yaml_file, 'r') as f:
        return yaml.safe_load(f)

def run_command(command, input_data=None):
    """Run a command with detailed logging."""
    print(f"Running command: {' '.join(command)}")
    try:
        result = subprocess.run(
            command,
            capture_output=True,
            text=True,
            check=True,
            input=input_data
        )
        print(f"Command output: {result.stdout}")
        return result.stdout.strip()
    except subprocess.CalledProcessError as e:
        print(f"Command failed with exit code {e.returncode}: {' '.join(command)}")
        print(f"Error output: {e.stderr}")
        raise
    except Exception as e:
        print(f"Unexpected error running command {' '.join(command)}: {str(e)}")
        raise

def get_token_and_host():
    """Get the token and host for API authentication."""
    print("\nGetting API token and host...")
    result = run_command([TB_COMMAND, "--output", "json", "info"])
    data = json.loads(result)
    token = data["local"]["token"]
    host = data["local"]["api"]
    print(f"Got token: {token[:10]}...")
    print(f"Got host: {host}")
    return token, host

def run_test_cases(endpoint, test_cases, base_url, token):
    """Run test cases for a specific endpoint."""
    print(f"\nTesting {endpoint} endpoint...")
    url = f"{base_url}/{endpoint}.ndjson"

    for test_case in test_cases:
        print(f"\nRunning test case: {test_case['name']}")

        # Prepare parameters
        params = {"token": token}
        if test_case.get("parameters"):
            params.update(dict(param.split("=") for param in test_case["parameters"].split("&")))

        # Make request
        print(f"Making request to {url} with params: {params}")
        response = requests.get(url, params=params)

        # Check HTTP status
        expected_status = test_case.get("expected_http_status", 200)
        assert response.status_code == expected_status, \
            f"Test '{test_case['name']}' failed: Expected status {expected_status}, got {response.status_code}"

        # Check response content
        if test_case["expected_result"]:
            if expected_status == 400:
                # For error responses, check the error message directly
                error_data = response.json()
                assert error_data["error"] == test_case["expected_result"], \
                    f"Test '{test_case['name']}' failed: Expected error '{test_case['expected_result']}', got '{error_data['error']}'"
            else:
                # For successful responses, parse as NDJSON
                expected_data = [json.loads(line) for line in test_case["expected_result"].strip().split('\n') if line.strip()]
                actual_data = [json.loads(line) for line in response.text.strip().split('\n') if line.strip()]
                print(f"Expected data: {expected_data}")
                print(f"Actual data: {actual_data}")
                assert actual_data == expected_data, \
                    f"Test '{test_case['name']}' failed: Expected {expected_data}, got {actual_data}"

def test_all_endpoints():
    """Test all endpoints using their respective YAML test files."""
    try:
        print("\n=== Setting up test environment ===")

        # Check if local server is running
        print("\nChecking local Tinybird server...")
        if not check_local_server():
            raise Exception("Local Tinybird server is not running. Please start it with 'docker compose up -d'")

        # Clear workspace
        print("\nClearing workspace...")
        try:
            run_command([TB_COMMAND, "workspace", "clear", "--yes"])
        except subprocess.CalledProcessError as e:
            print(f"Warning: Failed to clear workspace: {e.stderr}")
            print("Continuing with test execution...")

        # Deploy the project
        print("\nDeploying project...")
        run_command([TB_COMMAND, "deploy"])

        # Send test data to Redpanda
        print("\nSending test data to Redpanda...")
        with open("fixtures/kafka_water_meters.ndjson", "r") as f:
            data = f.read()
            run_command(["docker", "exec", "-i", "redpanda", "rpk", "topic", "produce", "water_metrics_demo", "-X", "brokers=redpanda:9092"], input_data=data)

        # Wait for data to be ingested
        print("\nWaiting for data ingestion...")
        time.sleep(45)  # Increased wait time

        print("\n=== Starting endpoint tests ===")

        # Get API token and host
        token, host = get_token_and_host()
        base_url = f"{host}/v0/pipes"

        # Find all YAML test files in the tests directory
        test_files = glob.glob("tests/*.yaml")

        # Run tests for each YAML file
        for test_file in test_files:
            # Extract endpoint name from filename (remove .yaml extension)
            endpoint = os.path.splitext(os.path.basename(test_file))[0]
            test_cases = load_test_cases(test_file)
            run_test_cases(endpoint, test_cases, base_url, token)

    finally:
        # Cleanup
        print("\n=== Cleaning up test environment ===")
        run_command([TB_COMMAND, "workspace", "clear", "--yes"])

### Enhance your CI/CD pipeline with local testing

To adapt the CI to use the real connection, an update to the existing CI workflow is needed. Modify `.github/workflows/tinybird-ci.yml`:

```yaml
name: Tinybird - Kafka full CI Workflow

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

env:
  TINYBIRD_HOST: ${{ secrets.TINYBIRD_HOST }}
  TINYBIRD_TOKEN: ${{ secrets.TINYBIRD_TOKEN }}

jobs:
  ci:
    runs-on: ubuntu-latest
    defaults:
      run:
        working-directory: '.'

    steps:
    - uses: actions/checkout@v4

    - name: Set up Python
      uses: actions/setup-python@v5
      with:
        python-version: '3.11'

    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install pytest requests pyyaml

    - name: Install Tinybird CLI
      run: |
        curl https://tinybird.co | sh

    - name: Set up Docker Compose
      run: |
        docker compose up -d
        # Wait for services to be ready
        sleep 30
        # Check if services are running
        docker compose ps
        # Check Redpanda logs
        docker compose logs redpanda
        # Check Tinybird logs
        docker compose logs tinybird-local

    - name: tb info
      run: |
        tb local status

    - name: create topic
      run: |
        docker exec -i redpanda rpk topic create water_metrics_demo -X brokers=redpanda:9092

    - name: Run tests
      env:
        TB_COMMAND: "tb"
        DOCKER_HOST: "unix:///var/run/docker.sock"
      run: |
        pytest tests/tests.py -v -s

    - name: Deployment check
      run: tb --cloud --host ${{ env.TINYBIRD_HOST }} --token ${{ env.TINYBIRD_TOKEN }} deploy --check