Aiven for Apache Kafka

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.clients.consumer.*;
import java.util.*;

// Producer Configuration
Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "kafka-service.aivencloud.com:12345");
producerProps.put("security.protocol", "SSL");
producerProps.put("ssl.truststore.location", "/path/to/client.truststore.jks");
producerProps.put("ssl.truststore.password", "changeit");
producerProps.put("ssl.keystore.location", "/path/to/client.keystore.p12");
producerProps.put("ssl.keystore.password", "changeit");
producerProps.put("ssl.key.password", "changeit");
producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);

ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "value");
producer.send(record);
producer.close();

// Consumer Configuration
Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "kafka-service.aivencloud.com:12345");
consumerProps.put("security.protocol", "SSL");
consumerProps.put("ssl.truststore.location", "/path/to/client.truststore.jks");
consumerProps.put("ssl.truststore.password", "changeit");
consumerProps.put("ssl.keystore.location", "/path/to/client.keystore.p12");
consumerProps.put("ssl.keystore.password", "changeit");
consumerProps.put("group.id", "my-consumer-group");
consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Collections.singletonList("my-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", 
            record.offset(), record.key(), record.value());
    }
}

const { Kafka } = require('kafkajs');
const fs = require('fs');

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['kafka-service.aivencloud.com:12345'],
  ssl: {
    rejectUnauthorized: true,
    ca: [fs.readFileSync('./ca.pem', 'utf-8')],
    key: fs.readFileSync('./service.key', 'utf-8'),
    cert: fs.readFileSync('./service.cert', 'utf-8')
  }
});

// Producer
const producer = kafka.producer();
await producer.connect();
await producer.send({
  topic: 'my-topic',
  messages: [
    { key: 'user-123', value: JSON.stringify({ event: 'login' }) }
  ]
});
await producer.disconnect();

// Consumer
const consumer = kafka.consumer({ groupId: 'my-consumer-group' });
await consumer.connect();
await consumer.subscribe({ topic: 'my-topic', fromBeginning: true });

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    console.log({
      partition,
      offset: message.offset,
      value: message.value.toString(),
    });
  },
});

package main

import (
    "crypto/tls"
    "crypto/x509"
    "github.com/Shopify/sarama"
    "io/ioutil"
    "log"
)

func main() {
    // Load certificates
    cert, err := tls.LoadX509KeyPair("service.cert", "service.key")
    if err != nil {
        log.Fatal(err)
    }

    caCert, err := ioutil.ReadFile("ca.pem")
    if err != nil {
        log.Fatal(err)
    }

    caCertPool := x509.NewCertPool()
    caCertPool.AppendCertsFromPEM(caCert)

    tlsConfig := &tls.Config{
        Certificates: []tls.Certificate{cert},
        RootCAs:      caCertPool,
    }

    // Producer
    config := sarama.NewConfig()
    config.Net.TLS.Enable = true
    config.Net.TLS.Config = tlsConfig
    config.Producer.Return.Successes = true

    producer, err := sarama.NewSyncProducer(
        []string{"kafka-service.aivencloud.com:12345"}, 
        config,
    )
    if err != nil {
        log.Fatal(err)
    }
    defer producer.Close()

    msg := &sarama.ProducerMessage{
        Topic: "my-topic",
        Value: sarama.StringEncoder("Hello from Go!"),
    }

    partition, offset, err := producer.SendMessage(msg)
    if err != nil {
        log.Fatal(err)
    }

    log.Printf("Message sent to partition %d at offset %d\n", partition, offset)
}