Kafka Connector for Infinispan

Introduction

This is a Kafka Connector to connect to an Infinispan instance (domain or standalone). For more information about Kafka Connect, take a look here

Running

mvn clean package

then in core folder

export CLASSPATH="$(find target/ -type f -name '*.jar'| grep '\-package' | tr '\n' ':')"

and you're now able to run from top folder

$KAFKA_HOME/bin/connect-standalone $KAFKA_HOME/config/connect-standalone.properties config/InfinispanSinkConnector.properties

Status

• Currently only the Sink Connector (from Kafka to Infinispan) has been developed and still need work.

Sink Connector Properties

Name	Description	Type	Default	Importance
infinispan.connection.hosts	List of comma separated Infinispan hosts	string	localhost	high
infinispan.connection.hotrod.port	Infinispan Hot Rod port	int	11222	high
infinispan.connection.cache.name	Infinispan Cache name of use	String	default	medium
infinispan.use.proto	If true, the Remote Cache Manager will be configured to use protostream schemas	boolean	false	medium
infinispan.proto.marshaller.class	If infinispan.use.proto is true, this option has to contain an annotated protostream class to be used	Class	String.class	medium
infinispan.cache.force.return.values	By default, previously existing values for Map operations are not returned, if set to true the values will be returned	boolean	false	low
infinispan.use.lifespan	If true, the Remote Cache Manager will be configured to use Lifespan associated to cache entries	boolean	false	low
infinispan.use.maxidle	If true, the Remote Cache Manager will be configured to use Max idle value associated to cache entries	boolean	false	low
infinispan.cache.lifespan.entry	If infinispan.use.lifespan is true, this option has to the lifespan associated with the entries to be stored (in seconds)	long	false	low
infinispan.cache.maxidle.entry	If infinispan.use.maxidle is true, this option has to the max idle associated with the entries to be stored (in seconds)	long	false	low
infinispan.hotrod.protocol.version	The infinispan hotrod client protocol version to use	String	DEFAULT_PROTOCOL_VERSION	low
infinispan.hotrod.socket_timeout	The infinispan hotrod client timeout for socket read/writes	int	5000	low
infinispan.hotrod.connect_timeout	The infinispan hotrod client timeout for connections	int	5000	low
infinispan.hotrod.max_retries	The infinispan hotrod client maximum number of operation retries	int	5	low

Configuration example

Suppose you want to store in your Infinispan cache object of kind Author. Here is Protostream annotated Author class:

package org.infinispan.kafka;

import java.io.Serializable;

import org.infinispan.protostream.annotations.ProtoDoc;
import org.infinispan.protostream.annotations.ProtoField;

@ProtoDoc("@Indexed")
public class Author implements Serializable {

   private String name;

   @ProtoField(number = 1, required = true)
   public String getName() {
      return name;
   }

   public void setName(String name) {
      this.name = name;
   }

   @Override
   public String toString() {
      return "Author [name=" + name + "]";
   }
}

You can then define the following configuration for your Infinispan Sink Connector

name=InfinispanSinkConnector
topics=mytopic
tasks.max=1
connector.class=org.infinispan.kafka.InfinispanSinkConnector
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter

infinispan.connection.hosts=127.0.0.1
infinispan.connection.hotrod.port=11222
infinispan.connection.cache.name=default
infinispan.cache.force.return.values=false
infinispan.use.proto=true
infinispan.proto.marshaller.class=org.infinispan.kafka.Author

At this point you will be able to run your connector. You'll need a running Infinispan server and a running Kafka server. In your Infinispan Kafka working directory run:

mvn clean package
export CLASSPATH="$(find target/ -type f -name '*.jar'| grep '\-package' | tr '\n' ':')"
$KAFKA_HOME/bin/connect-standalone $KAFKA_HOME/config/connect-standalone.properties config/InfinispanSinkConnector.properties

At this point you can try to send some messages to Kafka:

package org.infinispan.kafka;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

public class SimpleProducer {

    public static void main(String[] args) throws JsonProcessingException {

        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());

        KafkaProducer<String, String> prod = new KafkaProducer<String, String>(props);
        Author author = new Author();
        author.setName("oscerd");
        
        ObjectMapper mapper = new ObjectMapper();

        prod.send(new ProducerRecord<String, String>("test", "key1", mapper.writeValueAsString(author)));

        prod.close();
    }
}

In your Infinispan default cache you should now have a key/value pair with key1 as key and an Author object as value.

Releasing

Make sure MAVEN_HOME/conf/settings.xml contains credentials for the release repository. Add the following section in <servers>:

<server>
   <id>jboss-snapshots-repository</id>
   <username>RELEASE_USER</username>
   <password>RELEASE_PASS</password>
</server>
<server>
   <id>jboss-releases-repository</id>
   <username>RELEASE_USER</username>
   <password>RELEASE_PASS</password>
</server>

• Run mvn release:prepare release:perform -B