Integrating OpenClaw Rating API with Apache Kafka for Real‑Time Feedback Pipelines

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.5.0</version>
</dependency>

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.15.2</version>
</dependency>

implementation "org.apache.kafka:kafka-clients:3.5.0"
implementation "com.fasterxml.jackson.core:jackson-databind:2.15.2"

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.Properties;

public class RatingProducer {
    private final KafkaProducer<String, String> producer;
    private final String topic = "openclaw.ratings";
    private final ObjectMapper mapper = new ObjectMapper();

    public RatingProducer(String bootstrapServers) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        this.producer = new KafkaProducer<>(props);
    }

    public void sendRating(Rating rating) throws Exception {
        String key = rating.getUserId(); // optional partitioning key
        String value = mapper.writeValueAsString(rating);
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                // Log and optionally retry
                System.err.println("Failed to send rating: " + exception.getMessage());
            } else {
                System.out.println("Rating sent to partition " + metadata.partition()
                                   + " offset " + metadata.offset());
            }
        });
    }

    public void close() {
        producer.flush();
        producer.close();
    }

    // Simple POJO for rating payload
    public static class Rating {
        private String userId;
        private String itemId;
        private int score; // 1‑5
        private String comment;
        private long timestamp;

        // getters & setters omitted for brevity
    }
}

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.net.http.*;
import java.net.URI;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class RatingConsumer {
    private final KafkaConsumer<String, String> consumer;
    private final ObjectMapper mapper = new ObjectMapper();
    private final HttpClient httpClient = HttpClient.newHttpClient();
    private final String openClawEndpoint = "https://api.openclaw.io/v1/ratings";
    private final String apiKey = System.getenv("OPENCLAW_API_KEY");

    public RatingConsumer(String bootstrapServers, String groupId) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        this.consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("openclaw.ratings"));
    }

    public void pollAndProcess() {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                try {
                    Rating rating = mapper.readValue(record.value(), Rating.class);
                    sendToOpenClaw(rating);
                } catch (Exception e) {
                    // Handle deserialization or API errors
                    System.err.println("Processing failed: " + e.getMessage());
                }
            }
            consumer.commitSync();
        }
    }

    private void sendToOpenClaw(Rating rating) throws Exception {
        String json = mapper.writeValueAsString(rating);
        HttpRequest request = HttpRequest.newBuilder()
                .uri(URI.create(openClawEndpoint))
                .header("Content-Type", "application/json")
                .header("Authorization", "Bearer " + apiKey)
                .POST(HttpRequest.BodyPublishers.ofString(json))
                .build();

        HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
        if (response.statusCode() != 201) {
            throw new RuntimeException("OpenClaw rejected rating: " + response.body());
        }
        System.out.println("Rating persisted, id: " + response.body());
    }

    // Reuse the same POJO as in the producer
    public static class Rating {
        private String userId;
        private String itemId;
        private int score;
        private String comment;
        private long timestamp;
        // getters & setters omitted
    }

    public static void main(String[] args) {
        RatingConsumer consumer = new RatingConsumer("localhost:9092", "rating-consumer-group");
        consumer.pollAndProcess();
    }
}

# Dockerfile
FROM eclipse-temurin:17-jre-alpine
WORKDIR /app
COPY target/rating-consumer.jar .
ENTRYPOINT ["java","-jar","rating-consumer.jar"]

# values.yaml
replicaCount: 3
image:
  repository: your-registry/rating-consumer
  tag: "1.0.0"
  pullPolicy: IfNotPresent
service:
  type: ClusterIP
  port: 8080
resources:
  limits:
    cpu: "500m"
    memory: "512Mi"
  requests:
    cpu: "250m"
    memory: "256Mi"
autoscaling:
  enabled: true
  minReplicas: 2
  maxReplicas: 10
  targetCPUUtilizationPercentage: 70

LOGGER.info("{\"event\":\"rating_sent\",\"userId\":\"{}\",\"itemId\":\"{}\",\"status\":\"success\"}", rating.getUserId(), rating.getItemId());