Intro
Couchbase Kafka Connector 1.2.0 just shipped. Along with the various bug fixes, there is new sample code for a Kafka consumer in addition to the Kafka producer that was previously available. To quickly review the terms:
- A Kafka producer writes data to Kafka, so it’s a source of messages from Kafka’s perspective.
- A consumer in Kafka terminology is a process that subscribes to topics and then does something with the feed of published messages that are emitted from a Kafka cluster. It’s basically a sink.
In this blog, you’ll get up and running with a “Hello World!”-style sample Kafka consumer that writes to Couchbase. Along the way, you’ll also get a sandbox environment with a Kafka broker and a single node Couchbase Server so that you can actually run and modify the sample consumer and producer.
Installing Prerequisites
El samples are part of the Couchbase Kafka Connector source tree. To get them, just clone the whole repository:
|
1 |
$ git clone git://github.com/Couchbase/Couchbase–kafka–connector.git /tmp/kafka–connector |
Now, let’s setup your testing environment using pre-configured Kafka and Couchbase Server images. You have to install Vagrant, VirtualBox, and Ansible in order to set them up locally. If you have these services installed somewhere else, make sure you adjust the host addresses throughout this guide appropriately.
|
1 |
$ cd /tmp/kafka–connector/env |
Check versions of dependencies:
|
1 2 3 |
$ ansible —version $ vboxmanage —version $ vagrant –v |
You can assign human readable names to the boxes by using the plugin for Vagrant. If you don’t already have it installed, use the following command:
|
1 |
$ vagrant complemento Instalar vagrant–hostsupdater |
Now you’re ready to provision the servers and get running:
$ vagrant upNote: If a server fails to install due to timeouts, retry “vagrant up” after a few minutes and it may work.
Verify that the hosts are responding:
|
1 2 |
$ ping couchbase1.vagrant $ ping kafka1.vagrant |
If you navigate to you should be able to see your single-node Couchbase Server configured with credentials Administrador/contraseña.
Building the Samples
To avoid any classpath issues, use maven to create a self-contained JAR file for each sample application.
The generator application is a minimal CLI application. It uses the Couchbase Java SDK to wrap input lines from STDIN into JSON documents and sends them to the “default” bucket on Couchbase Server:
|
1 2 |
$ cd /tmp/kafka–connector/samples/generator $ mvn assembly:assembly |
The producer attaches to Couchbase Server and transmits all mutations to Kafka. This application uses the couchbase-kafka-connector project behind the scenes.
|
1 2 |
$ cd /tmp/kafka–connector/samples/producer $ mvn assembly:assembly |
Consumer is a typical Kafka consumer, which by default just outputs any incoming message in the topic “default” to STDOUT.
|
1 2 |
$ cd /tmp/kafka–connector/samples/consumer $ mvn assembly:assembly |
Running the Samples
Now that you have everything prepared, it’s time to run all the samples. You’ll need three different shell sessions because each of them runs a process until stopped. We’ll assume that you are in the /tmp/kafka-connector/samples directory.
First, start your generator:
|
1 |
$ Java –jar generator/target/kafka–samples–generator–1.0–SNAPSHOT–jar–con–dependencies.jar |
It should output the connection settings and then fall to a command prompt El resultado de tu pregunta se muestra a continuación:. You can type anything there and verify that it’s being created properly by looking in the Couchbase Server Admin UI:
Documents from generator in the bucket
|
1 2 3 4 |
... INFO: Opened cubeta default El resultado de tu pregunta se muestra a continuación: hello, kafka demo! >> key=key–5, valor={“line”:“hello, kafka demo!”} |
At this moment you can run the connector example
|
1 |
$ Java –jar producer/target/kafka–samples–producer–1.0–SNAPSHOT–jar–con–dependencies.jar |
For every line you type in the generator, you will see a line from the producer like this:
|
1 |
RECEIVED: com.couchbase.kafka.DCPEvent@4e44cb88 |
The sample writes it just before sending the payload to Kafka, in the filter class implementation. Let’s check how the Kafka receives these messages.
|
1 2 3 |
$ Java –jar consumer/target/kafka–samples–consumer–1.0–SNAPSHOT–jar–con–dependencies.jar 1: {“line”:“hello, kafka demo!”} 2: {“line”:“hello, this is a test”} |
You can continue playing with it as long as all three services are running.
Developing with Couchbase Kafka Connector
Let’s move on and take a look at the code. All three applications are pretty friendly for experiments, for example, the generator fits in just a few lines:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
público clase Ejemplo { público estático vacío main(Cadena args[]) throws IOException { Random random = nuevo Random(); Clúster clúster = CouchbaseCluster.crear(“couchbase1.vagrant”); Cubo cubeta = clúster.openBucket(); BufferedReader input = nuevo BufferedReader(nuevo InputStreamReader(System.en)); Cadena línea; do { System.out.imprimir(“> “); línea = input.readLine(); si (línea == nulo) { break; } Cadena key = “key-“ + random.nextInt(10); ObjetoJson valor = ObjetoJson.crear().poner(“line”, línea); cubeta.actualizar o insertar(JsonDocument.crear(key, valor)); System.out.printf(“>> key=%s, value=%sn”, key, valor); } while (true); } } |
Basically, generator opens a connection to bucket “default” on your “couchbase1.vagrant” instance and writes your messages to random keys. You can extend it to send other types of events. Another thing you might want to try to do is to remove keys.
By default, Couchbase Connector for Kafka runs in server mode, where it borrows active thread and actively listens to Couchbase Server for new events. There are several points where you can apply your ideas or changes. The most obvious one is configuration builder, where you not only specify the credentials and addresses of the services you are connecting to, but you can also specify various serializer and filter classes.
The sample application implements several of them. Filter class is the simplest:
|
1 2 3 4 5 6 7 |
público clase SampleFilter implementa Filter { @Override público boolean pass(DCPEvent dcpEvent) { System.out.println(“RECEIVED: “ + dcpEvent); regresar true; } } |
Here you can put in any custom checks you want, and if pass() returns false, the connector discards the message and won’t send it on to Kafka.
Default Encoder, which comes with the connector distribution, tries to represent every message as JSON, but that probably is not what you need, so you can apply and conversion to DCPEvent instance and return byte array, which will be stored in Kafka. In this example, we just convert events to their string representation.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
público clase SampleEncoder extiende AbstractEncoder { público SampleEncoder(final VerifiableProperties properties) { super(properties); } @Override público byte[] toBytes(final DCPEvent dcpEvent) { si (dcpEvent.message() instanceof MutationMessage) { MutationMessage message = (MutationMessage) dcpEvent.message(); regresar message.contenido().toString(CharsetUtil.UTF_8).getBytes(); } otro { regresar dcpEvent.message().toString().getBytes(); } } } |
A more advanced setting is StateSerializer interface. By implementing it, you can control how the library will track stream cursors (i.e. the sequence numbers for every partition inside Couchbase Server), and whether it will resume after connector restart. There is a Zookeeper implementation of state serializer in the distribution. Here in the sample, we’ve implemented NullStateSerializer which doesn’t persist anything, but it does show a minimal implementation.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
público clase NullStateSerializer implementa StateSerializer { público NullStateSerializer(final CouchbaseKafkaEnvironment environment) { } @Override público vacío dump(BucketStreamAggregatorState aggregatorState) { } @Override público vacío dump(BucketStreamAggregatorState aggregatorState, short partition) { } @Override público BucketStreamAggregatorState cargar(BucketStreamAggregatorState aggregatorState) { regresar nuevo BucketStreamAggregatorState(aggregatorState.name()); } @Override público BucketStreamState cargar(BucketStreamAggregatorState aggregatorState, short partition) { regresar nuevo BucketStreamState(partition, 0, 0, 0xffffffff, 0, 0xffffffff); } } |
The last component of your demo cluster is the Kafka consumer AbstractConsumer, which is a pretty typical instance of a consumer. It consists of two parts: , which implements bootstrap and positioning on the Kafka topic, and PrintConsumer, which carries your “business logic”, or just outputs every message it gets passed by AbstractConsumer:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
público clase PrintConsumer extiende AbstractConsumer { público PrintConsumer(Cadena[] seedBrokers, int port) { super(seedBrokers, port); } público PrintConsumer(Cadena seedBroker, int port) { super(seedBroker, port); } @Override público vacío handleMessage(long offset, byte[] bytes) { System.out.println(Cadena.valueOf(offset) + “: “ + nuevo Cadena(bytes)); } } |
As in the other examples here, you can play around with modifying the sample consumer. You can even close the circuit by sending everything back to Couchbase Server. Kafka is distributed software, just like Couchbase Server, so keep that in mind when running on your own cluster and adjust the main() function accordingly. In our sample, we have only a single partition, partition (0) in Kafka, so our main looks like this:
|
1 2 3 4 5 6 |
público clase Ejemplo { público estático vacío main(Cadena args[]) { PrintConsumer example = nuevo PrintConsumer(“kafka1.vagrant”, 9092); example.run(“default”, 0); } } |
Of course, in a production cluster you’ll be running more than one partition.
Conclusión
I hope this helps you get off to a good start with Couchbase and Kafka. Cheers!
Deja un comentario
Lo siento, debes estar conectado para publicar un comentario.