Confluent Platform

# Download Confluent Platform
curl -O http://packages.confluent.io/archive/7.5/confluent-7.5.0.tar.gz
tar -xzf confluent-7.5.0.tar.gz
cd confluent-7.5.0

# Iniciar todos os serviços
bin/confluent local services start

# Ou iniciar serviços individuais
bin/confluent local services kafka start
bin/confluent local services schema-registry start
bin/confluent local services ksql-server start
bin/confluent local services control-center start

# schema-registry.properties
listeners=http://0.0.0.0:8081
kafkastore.bootstrap.servers=localhost:9092
kafkastore.topic=_schemas
debug=false

# Registrar schema via API
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"name\":\"id\",\"type\":\"int\"},{\"name\":\"name\",\"type\":\"string\"}]}"}' \
  http://localhost:8081/subjects/user-value/versions

-- Criar stream a partir de tópico Kafka
CREATE STREAM user_events (
    user_id INT,
    event_type STRING,
    timestamp BIGINT
) WITH (
    KAFKA_TOPIC='user-events',
    VALUE_FORMAT='JSON'
);

-- Criar table agregada
CREATE TABLE user_event_counts AS
SELECT 
    user_id,
    COUNT(*) as event_count,
    COLLECT_LIST(event_type) as event_types
FROM user_events
WINDOW TUMBLING (SIZE 1 HOUR)
GROUP BY user_id;

-- Query contínua
SELECT * FROM user_event_counts
WHERE event_count > 100;

{
  "name": "jdbc-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:postgresql://localhost:5432/mydb",
    "connection.user": "postgres",
    "connection.password": "password",
    "table.whitelist": "users,orders",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "topic.prefix": "postgres-",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://localhost:8081",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://localhost:8081"
  }
}

import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.kafka.clients.producer.*;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.GenericData;

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", KafkaAvroSerializer.class);
props.put("value.serializer", KafkaAvroSerializer.class);
props.put("schema.registry.url", "http://localhost:8081");

Producer producer = new KafkaProducer<>(props);

// Criar record Avro
GenericRecord user = new GenericData.Record(userSchema);
user.put("id", 123);
user.put("name", "John Doe");

ProducerRecord record = 
    new ProducerRecord<>("users", "user-123", user);

producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        exception.printStackTrace();
    } else {
        System.out.println("Message sent to " + metadata.topic() + 
                          " partition " + metadata.partition() + 
                          " offset " + metadata.offset());
    }
});

producer.close();

# Gerenciar serviços Confluent
confluent local services start
confluent local services stop
confluent local services status

# Schema Registry CLI
kafka-avro-console-producer --broker-list localhost:9092 \
  --topic users --property schema.registry.url=http://localhost:8081 \
  --property value.schema.id=1

# ksqlDB CLI
ksql http://localhost:8088

# Control Center
# Acesse http://localhost:9021

# Kafka Connect
curl -X GET http://localhost:8083/connectors
curl -X POST -H "Content-Type: application/json" \
  --data @connector-config.json \
  http://localhost:8083/connectors