Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/amitsaxena098/OpenKnowledgeStream/llms.txt

Use this file to discover all available pages before exploring further.

OpenKnowledgeStream manages Kafka entirely through hand-configured Java Properties objects rather than Spring Boot’s auto-configuration. The wiki-change-stream module constructs a KafkaProducer inside KafkaPublish, while the opensearch-wiki-indexer module constructs a KafkaConsumer inside KafkaConsume. Both connect to the same broker and share the same topic, so any broker address change must be applied in both classes.

​
Producer configuration

The producer is created in KafkaPublish’s constructor and publishes Change objects serialized as JSON.
​
bootstrap.servers
string
default:"localhost:9092"
Comma-separated list of Kafka broker addresses the producer uses to bootstrap its connection to the cluster. Change this to match your Kafka broker host and port.
​
key.serializer
string
Serializer class for the record key. Keys are plain strings, so StringSerializer is used.
​
value.serializer
string
Serializer class for the record value. Spring Kafka’s JsonSerializer converts each Change POJO to a JSON byte array before sending.
​
topic
string
default:"recent_change_stream"
The Kafka topic to which change events are published. This value is hardcoded as the topic field in KafkaPublish.
KafkaPublish.java
package WikiChangeStream.publish;

import Wikicommon.models.Change;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.stereotype.Component;

import java.util.Properties;

@Component
@Slf4j
public class KafkaPublish {
    private final String topic = "recent_change_stream";
    KafkaProducer<String, Change> producer = null;

    KafkaPublish() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("key.serializer", StringSerializer.class.getName());
        properties.put("value.serializer", JsonSerializer.class.getName());
        producer = new KafkaProducer<>(properties);
    }

    public void publish(Change recentChange) {
        ProducerRecord<String, Change> record = new ProducerRecord<>(topic, recentChange);

        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if(exception == null) {
                    log.info("Change published with title: {}", recentChange.getTitle());
                } else {
                    log.error("Exception occurred while publish: {}", exception.getMessage());
                }
            }
        });
    }
}

​
Consumer configuration

The consumer is created in KafkaConsume’s constructor and subscribes to recent_change_stream to receive Change events for indexing into OpenSearch.
​
bootstrap.servers
string
default:"localhost:9092"
Kafka broker address the consumer uses to bootstrap its connection. Must match the broker address used by the producer.
​
key.deserializer
string
Deserializer class for the record key. Matches the StringSerializer used on the producer side.
​
value.deserializer
string
Deserializer class for the record value. Spring Kafka’s JsonDeserializer reconstructs Change POJOs from the JSON byte arrays produced by JsonSerializer.
​
group.id
string
default:"wiki-indexer"
The consumer group ID. All consumer instances sharing this ID will cooperatively consume partitions of recent_change_stream.
​
auto.offset.reset
string
default:"earliest"
Determines where to start reading when no committed offset exists for the consumer group. earliest means the consumer will read from the beginning of the topic on first start.
​
JsonDeserializer.TRUSTED_PACKAGES
string
default:"*"
Comma-separated list of packages that JsonDeserializer is permitted to deserialize into. The wildcard * allows all packages, including the Wikicommon.models package where Change lives.
​
topic subscription
string
default:"recent_change_stream"
The topic the consumer subscribes to. Set via consumer.subscribe(Arrays.asList("recent_change_stream")) in the constructor.
KafkaConsume.java
package WikiIndexer.consumer;

import WikiIndexer.index.OpensearchIndexer;
import Wikicommon.models.Change;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

@Component
@Slf4j
public class KafkaConsume {

    private final OpensearchIndexer opensearchIndexer;
    KafkaConsumer<String, Change> consumer;

    KafkaConsume(OpensearchIndexer opensearchIndexer) {
        this.opensearchIndexer = opensearchIndexer;
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("key.deserializer", StringDeserializer.class.getName());
        properties.put("value.deserializer", JsonDeserializer.class.getName());
        properties.put("group.id", "wiki-indexer");
        properties.put("auto.offset.reset", "earliest");
        properties.put(
                JsonDeserializer.TRUSTED_PACKAGES,
                "*"
        );
        consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Arrays.asList("recent_change_stream"));
    }

    @Scheduled(fixedRate = 5000)
    public void consume() throws Exception {
        log.info("Starting kafka consumer....");

        ConsumerRecords<String, Change> record = consumer.poll(Duration.ofMillis(1000));
        for(ConsumerRecord<String, Change> change : record) {
            opensearchIndexer.index(change.value());
        }
    }
}

​
Changing the broker address

bootstrap.servers is hardcoded in two separate files. When pointing OpenKnowledgeStream at a non-local Kafka cluster, update the property in both locations:
  • wiki-change-stream/src/main/java/WikiChangeStream/publish/KafkaPublish.java
  • opensearch-wiki-indexer/src/main/java/WikiIndexer/consumer/KafkaConsume.java
Replace "localhost:9092" with your broker’s host and port, e.g. "kafka.example.com:9092".

Build docs developers (and LLMs) love

Get started for freeTalk to us