Start reading Kafka topic from specific Offset in Apache Camel

6
votes

I read all the documentation of Camel Kafka and the only approach I read is this one from git and the route builder that specifies 

    public void configure() throws Exception {
                    from("kafka:" + TOPIC
                                 + "?groupId=A"
                                 + "&autoOffsetReset=earliest"             // Ask to start from the beginning if we have unknown offset
                                 + "&consumersCount=2"                     // We have 2 partitions, we want 1 consumer per partition
                                 + "&offsetRepository=#offset")            // Keep the offset in our repository
                            .to("mock:result");

}

But for order for the Clients I need to use the Spring so my endpoint for kafka is this 

<!--DEFINE KAFKA'S TOPCIS AS ENDPOINT-->
        <endpoint id="tagBlink" uri="kafka:10.0.0.165:9092">
            <property key="topic" value="tagBlink"/>
            <property key="brokers" value="10.0.0.165:9092"/>
            <property key="offsetRepository" value="100"/>
        </endpoint>

But getting an exception

Could not find a suitable setter for property: offsetRepository as there isn't a setter method with same type: java.lang.String nor type conversion possible: No type converter available to convert from type: java.lang.String to the required type: org.apache.camel.spi.StateRepository with value 100

Is this possible with my current configuration? How can I resume from an specific offset? ?

2
javaapache-kafkaapache-camel

2 Answers

1
votes

After this time I managed to work with this. I followed the Spring Bean creation for this and checking the documentation for the FileStateRepository I need a File so I created a File Bean and added as constructor-arg. After that I added an init-method="doStart". This method load a file if exist, and if not it will create the file.

     <endpoint id="event" uri="kafka:localhost:9092">
        <property key="topic" value="eventTopic4"/>
        <property key="brokers" value="localhost:9092"/>
        <property key="autoOffsetReset" value="earliest"/>
        <property key="offsetRepository" value="#myRepo2"/>
    </endpoint>

    <bean id="myFileOfMyRepo" class="java.io.File">
        <constructor-arg type="java.lang.String" value="C:\repoDat\repo.dat"/>
    </bean>

    <bean id="myRepo2" class="org.apache.camel.impl.FileStateRepository " factory-method="fileStateRepository" init-method="doStart">
        <constructor-arg ref="myFileOfMyRepo"/>
    </bean>

After this I saw the Code of the KafkaConsumer of the Camel in Git. 

    offsetRepository.getState(serializeOffsetKey(topicPartition));
    if (offsetState != null && !offsetState.isEmpty()) {
        // The state contains the last read offset so you need to seek from the next one
        long offset = deserializeOffsetValue(offsetState) + 1;
        log.debug("Resuming partition {} from offset {} from state", topicPartition.partition(), offset);
        consumer.seek(topicPartition, offset);
    }

With this I managed to read from the last offset. I hope Camel Documentation add this extra steps for Kafka.

-1
votes

The important word is "repository" and not "offset": its not an integer value but it's a reference to a bean specifying WHERE the offset is persisted.

(Non-Spring)Example

// Create the repository in which the Kafka offsets will be persisted
FileStateRepository repository = FileStateRepository.fileStateRepository(new File("/path/to/repo.dat"));

// Bind this repository into the Camel registry
JndiRegistry registry = new JndiRegistry();
registry.bind("offsetRepo", repository);

// Configure the camel context
DefaultCamelContext camelContext = new DefaultCamelContext(registry);
camelContext.addRoutes(new RouteBuilder() {
    @Override
    public void configure() throws Exception {
        from("kafka:" + TOPIC + "?brokers=localhost:{{kafkaPort}}" +
                     "&groupId=A" +                            //
                     "&autoOffsetReset=earliest" +             // Ask to start from the beginning if we have unknown offset
                     "&offsetRepository=#offsetRepo")          // Keep the offsets in the previously configured repository
                .to("mock:result");
    }
});