STREAMING & EVENT GUIDES • APACHE KAFKA BROKER

Ghid Testare, Securizare și Utilizare Apache Kafka pe VPS

Acest ghid descrie modul în care verifici, administrezi și conectezi serviciul de streaming / mesagerie Apache Kafka (deja instalat pe VPS-ul 63.250.40.142) la aplicațiile tale (Java, Python, Node.js).

1. Verificarea Statusului Apache Kafka pe VPS

Conectează-te la VPS prin SSH:
```
ssh root@63.250.40.142
```

Verifică starea serviciului Kafka (și Zookeeper dacă Kafka este configurat în modul clasic):

systemctl status kafka
# Dacă Kafka folosește Zookeeper, verifică și statusul acestuia:
systemctl status zookeeper

Dacă serviciul nu este activ, pornește-l și activează-l la boot:
```
systemctl start kafka
systemctl enable kafka
```

2. Testare Interactivă din Linia de Comandă (CLI)

Utilitarele Kafka de CLI se află de obicei în directorul de instalare (ex: /opt/kafka/bin sau /usr/local/kafka/bin). Navighează în director:

cd /opt/kafka/bin  # Ajustează calea în funcție de locația instalării de pe VPS

A. Creează un Topic de Test

Vom crea un topic numit test-topic pe portul implicit 9092:

./kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

B. Listează Topic-urile Existente

./kafka-topics.sh --list --bootstrap-server localhost:9092

C. Pornește un Producer (Producător de mesaje)

./kafka-console-producer.sh --topic test-topic --bootstrap-server localhost:9092

Tastați câteva mesaje și apăsați Enter:

Hello Alin din Kafka!
Primul mesaj de test pe VPS.

(Pentru a ieși din producer apăsați Ctrl + C).

D. Pornește un Consumer (Consumator de mesaje)

Într-o altă sesiune de terminal, citim mesajele de la începutul topicului (--from-beginning):

./kafka-console-consumer.sh --topic test-topic --from-beginning --bootstrap-server localhost:9092

3. Conectarea Kafka la Aplicațiile Tale

Portul standard pentru traficul intern Kafka pe VPS este 9092.

A. Java Spring Boot (application.properties)

Adaugă dependența spring-kafka în pom.xml:

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=group_id_demo
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

B. Python FastAPI / Script (kafka-python)

Instalează biblioteca kafka-python (pip install kafka-python):

Producer Python:

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

producer.send('test-topic', {'mesaj': 'Hello din Python FastAPI!'})
producer.flush()
print("Mesaj trimis pe Kafka!")

Consumer Python:

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'test-topic',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    print(f"Mesaj primit: {message.value}")

C. Node.js Express (kafkajs)

Instalează pachetul kafkajs (npm install kafkajs):

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

const kafka = new Kafka({
  clientId: 'node-demo-app',
  brokers: ['localhost:9092']
});

const producer = kafka.producer();

async function runProducer() {
  await producer.connect();
  await producer.send({
    topic: 'test-topic',
    messages: [
      { value: 'Hello din Node.js Express via KafkaJS!' },
    ],
  });
  console.log("Mesaj trimis cu succes!");
  await producer.disconnect();
}

runProducer();

4. Comenzi Utile de Administrare Kafka

Afișează detaliile unui Topic specific (partiții, replici):

./kafka-topics.sh --describe --topic test-topic --bootstrap-server localhost:9092

Vezi grupurile de consumatori (Consumer Groups):

./kafka-consumer-groups.sh --list --bootstrap-server localhost:9092

Șterge un Topic de test:

./kafka-topics.sh --delete --topic test-topic --bootstrap-server localhost:9092