Mock GRPC Services for Unit Testing

On our day to day work we develop applications that include interactions with software components through I/O. Can be a database, a broker or some form of blob storage. Take for example the Cloud Components you interact with: Azure Storage Queue, SQS, Pub/Sub. The communication with those components usually happens with an SDK.

From the start testing will kick in, thus the interaction with those components should be tackled in a testing context. An approach is to use installations (or simulators) of those components and have the code interacting with an actual instance, just like the way it can be achieved by using test containers or by creating infrastructure for testing purposes only.
Another approach is to spin up a mock service of the components and have the tests interacting with it. A good example of this can be Hoverfly. A simulated http service is run during testing and test cases interact with it.
Both can be used on various situations depending on the qualities our testing process requires. We shall focus on the second approach applied on gRPC.

It is well known that most Google Cloud Components come with a gRPC api.
In our scenario we have an application publishing messages to Pub/Sub.

Let’s put the needed dependencies first

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>com.google.cloud</groupId>
                <artifactId>libraries-bom</artifactId>
                <version>24.1.2</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <dependencies>
        <dependency>
            <groupId>com.google.cloud</groupId>
            <artifactId>google-cloud-pubsub</artifactId>
        </dependency>
        <dependency>
            <groupId>io.grpc</groupId>
            <artifactId>grpc-testing</artifactId>
            <version>1.43.2</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.google.api.grpc</groupId>
            <artifactId>grpc-google-cloud-pubsub-v1</artifactId>
            <version>1.97.1</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

Let’s start with our publisher class.

package com.egkatzioura.notification.publisher;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;

import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutureCallback;
import com.google.api.core.ApiFutures;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;

public class UpdatePublisher {

    private final Publisher publisher;
    private final Executor executor;

    public UpdatePublisher(Publisher publisher, Executor executor) {
        this.publisher = publisher;
        this.executor = executor;
    }

    public CompletableFuture<String> update(String notification) {
        PubsubMessage pubsubMessage = PubsubMessage.newBuilder()
                                                           .setData(ByteString.copyFromUtf8(notification))
                                                                   .build();
        ApiFuture<String> apiFuture = publisher.publish(pubsubMessage);

        return toCompletableFuture(apiFuture);
    }

    private CompletableFuture<String> toCompletableFuture(ApiFuture<String> apiFuture) {
        final CompletableFuture<String> responseFuture = new CompletableFuture<>();

        ApiFutures.addCallback(apiFuture, new ApiFutureCallback<>() {
            @Override
            public void onFailure(Throwable t) {
                responseFuture.completeExceptionally(t);
            }

            @Override
            public void onSuccess(String result) {
                responseFuture.complete(result);
            }

        }, executor);
        return responseFuture;
    }

}

The publisher will send messages and return the CompletableFuture of the message Id sent.
So let’s test this class. Our goal is to sent a message and get the message id back. The service to mock and simulate is PubSub.
For this purpose we added the grpc api dependency on maven

        <dependency>
            <groupId>com.google.api.grpc</groupId>
            <artifactId>grpc-google-cloud-pubsub-v1</artifactId>
            <version>1.97.1</version>
            <scope>test</scope>
        </dependency>

We shall mock the api for publishing actions. The class to implement is PublisherGrpc.PublisherImplBase.

package com.egkatzioura.notification.publisher;

import java.util.UUID;

import com.google.pubsub.v1.PublishRequest;
import com.google.pubsub.v1.PublishResponse;
import com.google.pubsub.v1.PublisherGrpc;

import io.grpc.stub.StreamObserver;

public class MockPublisherGrpc extends PublisherGrpc.PublisherImplBase {

    private final String prefix;

    public MockPublisherGrpc(String prefix) {
        this.prefix = prefix;
    }

    @Override
    public void publish(PublishRequest request, StreamObserver<PublishResponse> responseObserver) {
        responseObserver.onNext(PublishResponse.newBuilder().addMessageIds(prefix+":"+UUID.randomUUID().toString()).build());
        responseObserver.onCompleted();
    }

}

As you see the message id will have a prefix we define.

This would be the PublisherGrpc implementation on the server side. Let us proceed to our unit test. The UpdatePublisher class can have a Publisher injected. This publisher will be configured to use the PublisherGrpc.PublisherImplBase created previously.

@Rule
public final GrpcCleanupRule grpcCleanup = new GrpcCleanupRule();

private static final String MESSAGE_ID_PREFIX = "message";

@Before
public void setUp() throws Exception {
String serverName = InProcessServerBuilder.generateName();

Server server = InProcessServerBuilder
.forName(serverName).directExecutor().addService(new MockPublisherGrpc(MESSAGE_ID_PREFIX)).build().start();

grpcCleanup.register(server);
...

Above we created a GRPC server that services in-process requests. Then we registered the mock service created previously.
Onwards we create the Publisher using that service and create an instance of the class to test.

...

private UpdatePublisher updatePublisher;

@Before
public void setUp() throws Exception {
String serverName = InProcessServerBuilder.generateName();

Server server = InProcessServerBuilder
.forName(serverName).directExecutor().addService(new MockPublisherGrpc(MESSAGE_ID_PREFIX)).build().start();

grpcCleanup.register(server);

ExecutorProvider executorProvider = testExecutorProvider();
ManagedChannel managedChannel = InProcessChannelBuilder.forName(serverName).directExecutor().build();

TransportChannel transportChannel = GrpcTransportChannel.create(managedChannel);
TransportChannelProvider transportChannelProvider = FixedTransportChannelProvider.create(transportChannel);

String topicName = "projects/test-project/topic/my-topic";
Publisher publisher = Publisher.newBuilder(topicName)
.setExecutorProvider(executorProvider)
.setChannelProvider(transportChannelProvider)
.build();

updatePublisher = new UpdatePublisher(publisher, Executors.newSingleThreadExecutor());
...

We pass a Channel to our publisher which points to our InProcessServer. Requests will be routed to the service we registered. Finally we can add our test.

@Test
public void testPublishOrder() throws ExecutionException, InterruptedException {
String messageId = updatePublisher.update("Some notification").get();
assertThat(messageId, containsString(MESSAGE_ID_PREFIX));
}

We did it! We created our in process gRPC Server in order to have tests for our gRPC driven services!

You can find the code on GitHub!

Advertisement

Testing with Hoverfly and Java Part 6: JSON and JsonPath matchers

Previously we used the XML and Xpath Hoverfly matchers.
On this blog we shall focus on rules that assist us with the data exchanged using Json.

The default Json matcher will compare the Json submitted with the Json expected. This means that the submitted Json shall be validated for all the elements and their value. New lines or any extra spaces as long as they don’t change the information that the JSON carries, will not prevent the request from being a success.

Let’s put our initial configuration that will make the Json match.

 


    @BeforeEach
    void setUp() {
        var simulation = SimulationSource.dsl(service("http://localhost:8085")
                .post("/json")
                .body(RequestFieldMatcher.newJsonMatcher("{\"document\":\"document-a\"}"))
                .willReturn(success(SUCCESS_RESPONSE, "application/json"))
                .post("/json/partial")
                .body(RequestFieldMatcher.newJsonPartialMatcher("{\"document\":\"document-a\"}"))
                .willReturn(success(SUCCESS_RESPONSE, "application/json"))
                .post("/jsonpath")
                .body(RequestFieldMatcher.newJsonPathMatch("$.document[1].description"))
                .willReturn(success(SUCCESS_RESPONSE, "application/json"))
        );

        var localConfig = HoverflyConfig.localConfigs().disableTlsVerification().asWebServer().proxyPort(8085);
        hoverfly = new Hoverfly(localConfig, SIMULATE);
        hoverfly.start();
        hoverfly.simulate(simulation);
    }

    @AfterEach
    void tearDown() {
        hoverfly.close();
    }

    

In our first example we will try to match the Json of our request with the Json expected.

    @Test
    void testJsonExactMatch() {
        var client = HttpClient.newHttpClient();

        var exactRequest = HttpRequest.newBuilder()
                .uri(URI.create("http://localhost:8085/json"))
                .POST(HttpRequest.BodyPublishers.ofString("   {\"document\":    \"document-a\"}"))
                .build();

        var exactResponse = client.sendAsync(exactRequest, HttpResponse.BodyHandlers.ofString())
                .thenApply(HttpResponse::body)
                .join();

        Assertions.assertEquals(SUCCESS_RESPONSE, exactResponse);
    }

Also let’s make sure there is going to be a failure on an extra element.

    @Test
    void testJsonNoMatch() {
        var client = HttpClient.newHttpClient();

        var exactRequest = HttpRequest.newBuilder()
                .uri(URI.create("http://localhost:8085/json"))
                .POST(HttpRequest.BodyPublishers.ofString("{\"doc2\":\"value\", \"document\":\"document-a\"}"))
                .build();

        var exactResponse = client.sendAsync(exactRequest, HttpResponse.BodyHandlers.ofString())
                .join();

        Assertions.assertEquals(502, exactResponse.statusCode());
    }

Now let’s see the non exact matcher.

    @Test
    void testJsonPartialMatch() {
        var client = HttpClient.newHttpClient();

        var exactRequest = HttpRequest.newBuilder()
                .uri(URI.create("http://localhost:8085/json/partial"))
                .POST(HttpRequest.BodyPublishers.ofString("{\"doc2\":\"value\", \"document\":\"document-a\"}"))
                .build();

        var exactResponse = client.sendAsync(exactRequest, HttpResponse.BodyHandlers.ofString())
                .thenApply(HttpResponse::body)
                .join();

        Assertions.assertEquals(SUCCESS_RESPONSE, exactResponse);
    }

So far we checked matching the whole payload. Let’s try the Jsonpath approach. The example below does match.

    @Test
    void testJsonPathMatch() {
        var client = HttpClient.newHttpClient();

        var exactRequest = HttpRequest.newBuilder()
                .uri(URI.create("http://localhost:8085/jsonpath"))
                .POST(HttpRequest.BodyPublishers.ofString("{\"document\":[{\"description\":\"description-1\"},{\"description\":\"description-2\"}]}"))
                .build();

        var exactResponse = client.sendAsync(exactRequest, HttpResponse.BodyHandlers.ofString())
                .thenApply(HttpResponse::body)
                .join();

        Assertions.assertEquals(SUCCESS_RESPONSE, exactResponse);
    }

But the example below won’t match


    @Test
    void testJsonPathNoMatch() {
        var client = HttpClient.newHttpClient();

        var exactRequest = HttpRequest.newBuilder()
                .uri(URI.create("http://localhost:8085/jsonpath"))
                .POST(HttpRequest.BodyPublishers.ofString("{\"document\":[{\"description\":\"description-1\"}]}"))
                .build();

        var exactResponse = client.sendAsync(exactRequest, HttpResponse.BodyHandlers.ofString())
                .join();

        Assertions.assertEquals(502, exactResponse.statusCode());
    }

That’s it we did use the Json and JsonPath matchers for the Json based data!

Testing with Hoverfly and Java Part 5: XML and Xpath matchers

Previously we worked with some of the existing Hoverfly matchers like the regex, glob and exact.
Each one serves its purpose but we might want some rules that assist us with the format of the data exchanged through our requests.

On this blog we will focus on the matchers for xml.

 

The default xml matcher will compare the xml submitted with the xml expected. This means that the submitted xml shall be validated node by node value by value. New lines or any extra spaces as long as they don’t change the content that the xml carries will not prevent the request from being a success.

Let’s put our initial configuration that will make the xml match.

	public static final String SUCCESS_RESPONSE = "<response>"
			+ "<result>success</result>"
			+ "</response>";

	private Hoverfly hoverfly;

	@BeforeEach
	void setUp() {
		var simulation = SimulationSource.dsl(service("http://localhost:8085")
				.post("/xml")
				.body(RequestFieldMatcher.newXmlMatcher("<document type=\"xml\">"
						+ "xml-request"
						+ "</document>"))
				.willReturn(success(SUCCESS_RESPONSE, "application/xml")));

		var localConfig = HoverflyConfig.localConfigs().disableTlsVerification().asWebServer().proxyPort(8085);
		hoverfly = new Hoverfly(localConfig, SIMULATE);
		hoverfly.start();
		hoverfly.simulate(simulation);
	}

	@AfterEach
	void tearDown() {
		hoverfly.close();
	}

So in our first example we will try to match the xml of our request with the xml expected.

	@Test
	void testXmlExactMatch() {
		var client = HttpClient.newHttpClient();

		var exactRequest = HttpRequest.newBuilder()
				.uri(URI.create("http://localhost:8085/xml"))
				.POST(HttpRequest.BodyPublishers.ofString("  <document type=\"xml\">\n\n"
						+ "xml-request"
						+ "</document>\t"))
				.build();

		var exactResponse = client.sendAsync(exactRequest, HttpResponse.BodyHandlers.ofString())
				.thenApply(HttpResponse::body)
				.join();

		Assertions.assertEquals(SUCCESS_RESPONSE, exactResponse);
	}

As you see regardless of the new lines and the tabs, our request will be successful since the xml data do match.

Now let’s try to add a node to the xml.

	@Test
	void testXmlNoMatch() {
		var client = HttpClient.newHttpClient();

		var exactRequest = HttpRequest.newBuilder()
				.uri(URI.create("http://localhost:8085/xml"))
				.POST(HttpRequest.BodyPublishers.ofString("  <document type=\"xml\">\n\n"
						+ "xml-request"
						+ "</document>\t<empty-node>ok</empty-node>"))
				.build();

		var exactResponse = client.sendAsync(exactRequest, HttpResponse.BodyHandlers.ofString())
				.join();

		Assertions.assertEquals(502, exactResponse.statusCode());
	}

The xml does not match thus it will fail.

Let’s focus to another problem. Since the data exchanged are dynamic, chances are that exact matches might not be possible. Also you might not need to focus on all the information submitted but just a specific section of the information exchanged. Therefore an XPath matcher becomes handy.

Will enhance the initial setup with an XPath rule.


	@BeforeEach
	void setUp() {
		var simulation = SimulationSource.dsl(service("http://localhost:8085")
				.post("/xml")
				.body(RequestFieldMatcher.newXmlMatcher("<document type=\"xml\">"
						+ "xml-request"
						+ "</document>"))
				.willReturn(success(SUCCESS_RESPONSE, "application/xml"))
				.post("/xpath")
				.body(RequestFieldMatcher.newXpathMatcher("/document/payment[amount=1]"))
				.willReturn(success(SUCCESS_RESPONSE, "application/xml"))
		);

		var localConfig = HoverflyConfig.localConfigs().disableTlsVerification().asWebServer().proxyPort(8085);
		hoverfly = new Hoverfly(localConfig, SIMULATE);
		hoverfly.start();
		hoverfly.simulate(simulation);
	}

If there is a document node with a payment node and the value on the amount node is 1 there will be a match
Let’s go for a positive scenario

	@Test
	void testXpathMatch() {
		var client = HttpClient.newHttpClient();

		var exactRequest = HttpRequest.newBuilder()
				.uri(URI.create("http://localhost:8085/xpath"))
				.POST(HttpRequest.BodyPublishers.ofString("  <document type=\"xml\">\n\n"
						+ "<payment><amount>142</amount></payment>"
						+ "<payment><amount>1</amount><currency>GBP</currency></payment>"
						+ "<payment>invalid</payment>"
						+ "</document>\t"))
				.build();

		var exactResponse = client.sendAsync(exactRequest, HttpResponse.BodyHandlers.ofString())
				.thenApply(HttpResponse::body)
				.join();

		Assertions.assertEquals(SUCCESS_RESPONSE, exactResponse);
	}

As expected we got a match.
Let’s go for a negative scenario.

	@Test
	void testXpathNoMatch() {
		var client = HttpClient.newHttpClient();

		var exactRequest = HttpRequest.newBuilder()
				.uri(URI.create("http://localhost:8085/xpath"))
				.POST(HttpRequest.BodyPublishers.ofString("  <document type=\"xml\">\n\n"
						+ "<payment><amount>142</amount></payment>"
						+ "<payment><amount>no-match</amount><currency>GBP</currency></payment>"
						+ "<payment>invalid</payment>"
						+ "</document>\t"))
				.build();

		var exactResponse = client.sendAsync(exactRequest, HttpResponse.BodyHandlers.ofString())
				.join();

		Assertions.assertEquals(502, exactResponse.statusCode());
	}

That’s it we did use the xml and xpath matchers for the xml based data. The next blog shall focus on the JSON based matchers.

Static Factories are Great!

Every now and then I jump on classes with multiple constructors or classes that are rigorous to work with. Let alone not being able to mock part of their components and at the end being forced to use reflection for testing (mockito based, old school, you choose).

Imagine a Producer class that you use for Kafka. A class that provides you with some abstraction on sending messages.

package com.gkatzioura.kafka.producer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

@Slf4j
public class StrMessageProducer {

	private Producer<String,String> producer;
	private String topic = "test-topic";
	
	StrMessageProducer() {
		var kafkaProperties = new Properties();
		kafkaProperties.put("bootstrap.servers",System.getProperty("bootstrap.servers"));
		kafkaProperties.put("key.serializer",System.getProperty("key.serializer"));
		kafkaProperties.put("value.serialize",System.getProperty("value.serializer"));
		var kafkaProducer = new KafkaProducer<String,String>(kafkaProperties);
		this.producer = kafkaProducer;
	}

	public void send(String message) {
		var producerRecord = new ProducerRecord<String,String>(topic,null, message);
		try {
			var metadata = producer.send(producerRecord).get();
			log.info("Submitted {}",metadata.offset());
		}
		catch (InterruptedException |ExecutionException e) {
			log.error("Could not send record",e);
		}
	}
}

Apart from being an ugly class it is also very hard to change some of its components.

For example

  • I cannot use this class to post to another topic
  • I cannot use this class to use a different server configuration apart from the one on the properties
  • It’s difficult to test the functionality of the class since crucial components are created through the constructor

It’s obvious that the constructor in this case serves the purpose on creating a Kafka producer based on the system properties. But the responsibility of the class is to use that producer in order send messages in a specific way. Thus I will move the creation of the Producer from the constructor. Also because we might want to swap the topic used, I will also inject the topic instead of having it hardcoded.
By doing so we encourage dependency injection. We make it easy to swap the ingredients of the class however the execution would be the same.

package com.gkatzioura.kafka.producer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

@Slf4j
public class StrMessageProducer {

	private final Producer<String,String> producer;
	private final String topic;

	StrMessageProducer(Producer<String,String> producer, String topic) {
		this.producer = producer;
		this.topic = topic;
	}

	public void send(String message) {
		var producerRecord = new ProducerRecord<String,String>(topic,null, message);
		try {
			var metadata = producer.send(producerRecord).get();
			log.info("Submitted {}",metadata.offset());
		}
		catch (InterruptedException |ExecutionException e) {
			log.error("Could not send record",e);
		}
	}
}

But we still need the producer to be created somehow. This is where the factory pattern kicks in.

We shall add static factories in order to have instances of the StrMessageProducer class with different configurations.
Let’s add two factory methods
The first factory method would be based on system properties and the second on environment variables.

package com.gkatzioura.kafka.producer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

@Slf4j
public class StrMessageProducer {

	private final Producer<String,String> producer;
	private final String topic;

	StrMessageProducer(Producer<String,String> producer, String topic) {
		this.producer = producer;
		this.topic = topic;
	}

	public void send(String message) {
		var producerRecord = new ProducerRecord<String,String>(topic,null, message);
		try {
			var metadata = producer.send(producerRecord).get();
			log.info("Submitted {}",metadata.offset());
		}
		catch (InterruptedException |ExecutionException e) {
			log.error("Could not send record",e);
		}
	}

	public static StrMessageProducer createFromSystemPros() {
		var kafkaProperties = new Properties();
		kafkaProperties.put("bootstrap.servers",System.getProperty("bootstrap.servers"));
		kafkaProperties.put("key.serializer",System.getProperty("key.serializer"));
		kafkaProperties.put("value.serialize",System.getProperty("value.serializer"));
		var kafkaProducer = new KafkaProducer<String,String>(kafkaProperties);
		return new MessageProducer(kafkaProducer, System.getProperty("main.topic"));
	}

	public static StrMessageProducer createFromEnv() {
		var kafkaProperties = new Properties();
		kafkaProperties.put("bootstrap.servers",System.getenv("BOOTSTRAP_SERVERS"));
		kafkaProperties.put("key.serializer",System.getenv("KEY_SERIALIZER"));
		kafkaProperties.put("value.serialize",System.getenv("VALUE_SERIALIZER"));
		var kafkaProducer = new KafkaProducer<String,String>(kafkaProperties);
		return new MessageProducer(kafkaProducer, System.getProperty("MAIN_TOPIC"));
	}
}

You already see the benefits. You have a clean class ready to use as it is and you have some factory methods for convenience. Eventually you can add more static factories, some of them might also have arguments, for example the topic.

Also we can go one step further when we want to have multiple classes of MessageProducers and we want to utilise an interface. So we are going to introduce the MessageProducer interface which our StrMessageProducer class will implement. Also we are going to put the static factories to the interface.

So this will be our interface with the static factories.

package com.gkatzioura.kafka.producer;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;

public interface MessageProducer {
	
	void send(String message);

	static MessageProducer createFromSystemPros() {
		var kafkaProperties = new Properties();
		kafkaProperties.put("bootstrap.servers",System.getProperty("bootstrap.servers"));
		kafkaProperties.put("key.serializer",System.getProperty("key.serializer"));
		kafkaProperties.put("value.serialize",System.getProperty("value.serializer"));
		var kafkaProducer = new KafkaProducer<String,String>(kafkaProperties);
		return new StrMessageProducer(kafkaProducer, System.getProperty("main.topic"));
	}

	static MessageProducer createFromEnv() {
		var kafkaProperties = new Properties();
		kafkaProperties.put("bootstrap.servers",System.getenv("BOOTSTRAP_SERVERS"));
		kafkaProperties.put("key.serializer",System.getenv("KEY_SERIALIZER"));
		kafkaProperties.put("value.serialize",System.getenv("VALUE_SERIALIZER"));
		var kafkaProducer = new KafkaProducer<String,String>(kafkaProperties);
		return new StrMessageProducer(kafkaProducer, System.getProperty("MAIN_TOPIC"));
	}

}

And this would be our new StrMessageProducer class.

package com.gkatzioura.kafka.producer;

import java.util.concurrent.ExecutionException;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

@Slf4j
public class StrMessageProducer implements MessageProducer {

	private final Producer<String,String> producer;
	private final String topic;

	StrMessageProducer(Producer<String,String> producer, String topic) {
		this.producer = producer;
		this.topic = topic;
	}

	@Override
	public void send(String message) {
		var producerRecord = new ProducerRecord<String,String>(topic,null, message);
		try {
			var metadata = producer.send(producerRecord).get();
			log.info("Submitted {}",metadata.offset());
		}
		catch (InterruptedException |ExecutionException e) {
			log.error("Could not send record",e);
		}
	}

}

Let’s check the benefits

  • We can have various implementations of a MessageProducer class
  • We can add as many factories we want that serve our purpose
  • We can easily test the MessageProducer implementation by passing mocks to the constructors
  • We keep our codebase cleaner

 

Kafka & Zookeeper for Development: Connecting Clients to the Cluster

Previously we achieved to have our Kafka brokers connect to a ZooKeeper ensemble. Also we brought down some brokers checked the election leadership and produced/consumed some messages.

For now we want to make sure that we will be able to connect to those nodes. The problem with connecting to the ensemble we created previously is that it is located inside the container network. When a client interacts with one of the brokers and receives the full list of the brokers he will receive a list of IPs not accessible to it.

So the initial handshake of a client will be successful but then the client will try to interact withs some unreachable hosts.

In order to tackle this we will have a combination of workarounds.

The first one would be to bind the port of each Kafka broker to a different local ip.

kafka-1 will be mapped to 127.0.0.1:9092
kafka-2 will be mapped to 127.0.0.2:9092
kafka-3 will be mapped to 127.0.0.3:9092

So let’s create the aliases of those addresses

sudo ifconfig lo0 alias 127.0.0.2
sudo ifconfig lo0 alias 127.0.0.3

Now it’s possible to do the ip binding. Let’s also put those entries to our /etc/hosts. By doing this, we achieve our local network and our docker network to be in agreement on which broker they should access.

127.0.0.1	kafka-1
127.0.0.2	kafka-2
127.0.0.3	kafka-3

The next step is also to change the KAFKA_ADVERTISED_LISTENERS on each broker. We will adapt this to the DNS entry of each broker. By setting KAFKA_ADVERTISED_LISTENERS the clients from the outside can correctly connect to it, to an address reachable to them and not an address through the internal network. Further explanations can be found on this blog.

  kafka-1:
    container_name: kafka-1
    image: confluent/kafka
    ports:
    - "127.0.0.1:9092:9092"
    volumes:
    - type: bind
      source: ./server1.properties
      target: /etc/kafka/server.properties
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    environment:
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka-1:9092"
  kafka-2:
    container_name: kafka-2
    image: confluent/kafka
    ports:
      - "127.0.0.2:9092:9092"
    volumes:
      - type: bind
        source: ./server2.properties
        target: /etc/kafka/server.properties
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    environment:
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka-2:9092"
  kafka-3:
    container_name: kafka-3
    image: confluent/kafka
    ports:
      - "127.0.0.3:9092:9092"
    volumes:
      - type: bind
        source: ./server3.properties
        target: /etc/kafka/server.properties
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    environment:
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka-3:9092"

We see the port binding change as well as the KAFKA_ADVERTISED_LISTENERS. Now let’s wrap everything together in our docker-compose

version: "3.8"
services:
  zookeeper-1:
    container_name: zookeeper-1
    image: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOO_MY_ID: "1"
      ZOO_SERVERS: server.1=0.0.0.0:2888:3888;2181 server.2=zookeeper-2:2888:3888;2181 server.3=zookeeper-3:2888:3888;2181
  zookeeper-2:
    container_name: zookeeper-2
    image: zookeeper
    ports:
      - "2182:2181"
    environment:
      ZOO_MY_ID: "2"
      ZOO_SERVERS: server.1=zookeeper-1:2888:3888;2181 server.2=0.0.0.0:2888:3888;2181 server.3=zookeeper-3:2888:3888;2181
  zookeeper-3:
    container_name: zookeeper-3
    image: zookeeper
    ports:
      - "2183:2181"
    environment:
      ZOO_MY_ID: "3"
      ZOO_SERVERS: server.1=zookeeper-1:2888:3888;2181 server.2=0.0.0.0:2888:3888;2181 server.3=zookeeper-3:2888:3888;2181
  kafka-1:
    container_name: kafka-1
    image: confluent/kafka
    ports:
    - "127.0.0.1:9092:9092"
    volumes:
    - type: bind
      source: ./server1.properties
      target: /etc/kafka/server.properties
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    environment:
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka-1:9092"
  kafka-2:
    container_name: kafka-2
    image: confluent/kafka
    ports:
      - "127.0.0.2:9092:9092"
    volumes:
      - type: bind
        source: ./server2.properties
        target: /etc/kafka/server.properties
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    environment:
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka-2:9092"
  kafka-3:
    container_name: kafka-3
    image: confluent/kafka
    ports:
      - "127.0.0.3:9092:9092"
    volumes:
      - type: bind
        source: ./server3.properties
        target: /etc/kafka/server.properties
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    environment:
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka-3:9092"

You can find more on Compose on the Developers Essential Guide to Docker Compose.

Last but not least you can find the code on github.

Testing with Hoverfly and Java Part 4: Exact, Glob and Regex Matchers

Previously we used Hoverfly among its state feature.
So far our examples have been close to an absolute request match, thus on this blog we will focus on utilising the matchers.
Having a good range of matchers is very important because most API interactions are dynamic and you can’t always predict the example. Imagine a JWT signature. You can match the body but the signature might change per environment.

 

There are three type of matchers available.

  • The exact matcher: The fields headers should be an exact match
  • The glob matcher: A match that gives the ability to using the `*`
  • The regex matcher: A matcher that requires you to search again on the internet on how to make a regex
  • XML matcher: This about matching the xml as XML node by node value by value
  • Xpath matcher: Match based on a value match through Xpath
  • JSON matcher: Match the json exactly
  • JSON partial matcher: Match if the json submitted contains the Json values specified
  • JSONPath matcher: Just like the xpath match based the on the json path submitted

Let’s start with the exact matcher.

public class ExactMatcherTests {

	private Hoverfly hoverfly;

	@BeforeEach
	void setUp() {
		var simulation = SimulationSource.dsl(service("http://localhost:8085")
				.get("/exact")
				.header("Origin", RequestFieldMatcher.newExactMatcher("internal-server"))
				.willReturn(success("{\"exact\":true}", "application/json")));

		var localConfig = HoverflyConfig.localConfigs().disableTlsVerification().asWebServer().proxyPort(8085);
		hoverfly = new Hoverfly(localConfig, SIMULATE);
		hoverfly.start();
		hoverfly.simulate(simulation);
	}

	@AfterEach
	void tearDown() {
		hoverfly.close();
	}

	@Test
	void testExactMatcherSuccess() {
		var client = HttpClient.newHttpClient();
		var exactRequest = HttpRequest.newBuilder()
				.uri(URI.create("http://localhost:8085/exact"))
				.header("Origin","internal-server")
				.build();

		var exactResponse = client.sendAsync(exactRequest, HttpResponse.BodyHandlers.ofString())
				.thenApply(HttpResponse::body)
				.join();

		Assertions.assertEquals("{\"exact\":true}", exactResponse);
	}

	@Test
	void testExactMatcherFailure() {
		var client = HttpClient.newHttpClient();
		var exactRequest = HttpRequest.newBuilder()
				.uri(URI.create("http://localhost:8085/exact"))
				.build();

		var exactResponse = client.sendAsync(exactRequest, HttpResponse.BodyHandlers.ofString())
				.join();

		Assertions.assertEquals(502, exactResponse.statusCode());
	}

}

The failures or success come based on wether the header was matching exactly or not.

We shall use the glob match for a request parameter.


public class GlobMatcher {

	private Hoverfly hoverfly;

	@BeforeEach
	void setUp() {
		var simulation = SimulationSource.dsl(service("http://localhost:8085")
				.get("/glob")
				.queryParam("userName", RequestFieldMatcher.newGlobMatcher("john*"))
				.willReturn(success("{\"glob\":true}", "application/json")));

		var localConfig = HoverflyConfig.localConfigs().disableTlsVerification().asWebServer().proxyPort(8085);
		hoverfly = new Hoverfly(localConfig, SIMULATE);
		hoverfly.start();
		hoverfly.simulate(simulation);
	}

	@AfterEach
	void tearDown() {
		hoverfly.close();
	}

	@Test
	void testGlobMatcherSuccess() {
		var client = HttpClient.newHttpClient();
		var exactRequest = HttpRequest.newBuilder()
				.uri(URI.create("http://localhost:8085/glob?userName=johnDoe"))
				.build();

		var exactResponse = client.sendAsync(exactRequest, HttpResponse.BodyHandlers.ofString())
				.thenApply(HttpResponse::body)
				.join();

		Assertions.assertEquals("{\"glob\":true}", exactResponse);
	}

	@Test
	void testGlobMatcherFailure() {
		var client = HttpClient.newHttpClient();
		var exactRequest = HttpRequest.newBuilder()
				.uri(URI.create("http://localhost:8085/glob?userName=nojohnDoe"))
				.build();

		var exactResponse = client.sendAsync(exactRequest, HttpResponse.BodyHandlers.ofString())
				.join();

		Assertions.assertEquals(502, exactResponse.statusCode());
	}

}

Last let’s head for the regex matcher. The regex matcher will just check for a capital letter: ([A-Z])\w+

public class RegexMatcherTests {

	private Hoverfly hoverfly;

	@BeforeEach
	void setUp() {
		var simulation = SimulationSource.dsl(service("http://localhost:8085")
				.post("/regex")
				.body(RequestFieldMatcher.newRegexMatcher("([A-Z])\\w+"))
				.willReturn(success("{\"regex\":true}", "application/json")));

		var localConfig = HoverflyConfig.localConfigs().disableTlsVerification().asWebServer().proxyPort(8085);
		hoverfly = new Hoverfly(localConfig, SIMULATE);
		hoverfly.start();
		hoverfly.simulate(simulation);
	}

	@AfterEach
	void tearDown() {
		hoverfly.close();
	}

	@Test
	void testRegexMatcherSuccess() {
		var client = HttpClient.newHttpClient();
		var exactRequest = HttpRequest.newBuilder()
				.uri(URI.create("http://localhost:8085/regex"))
				.POST(HttpRequest.BodyPublishers.ofString("Contains capital letter"))
				.build();

		var exactResponse = client.sendAsync(exactRequest, HttpResponse.BodyHandlers.ofString())
				.thenApply(HttpResponse::body)
				.join();

		Assertions.assertEquals("{\"regex\":true}", exactResponse);
	}

	@Test
	void testRegexMatcherFailure() {
		var client = HttpClient.newHttpClient();
		var exactRequest = HttpRequest.newBuilder()
				.uri(URI.create("http://localhost:8085/regex"))
				.POST(HttpRequest.BodyPublishers.ofString("won't match due to capital letter missing"))
				.build();

		var exactResponse = client.sendAsync(exactRequest, HttpResponse.BodyHandlers.ofString())
				.join();

		Assertions.assertEquals(502, exactResponse.statusCode());
	}

}

That’s it we did use the basic matchers for exact, glob, and regex based. The next blog shall focus on the xml based matchers.

Kafka & Zookeeper for Development: Connecting Brokers to the Ensemble

Previously we created successfully a Zookeeper ensemble, now it’s time to add some Kafka brokers that will connect to the ensemble and we shall execute some commands.

We will pick up from the same docker compose file we compiled previously. First let’s jump on the configuration that a Kafka broker needs.

offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
group.initial.rebalance.delay.ms=0
socket.send.buffer.bytes=102400
delete.topic.enable=true
socket.request.max.bytes=104857600
log.cleaner.enable=true
log.retention.check.interval.ms=300000
log.retention.hours=168
num.io.threads=8
broker.id=0
log4j.opts=-Dlog4j.configuration=file:/etc/kafka/log4j.properties
log.dirs=/var/lib/kafka
auto.create.topics.enable=true
num.network.threads=3
socket.receive.buffer.bytes=102400
log.segment.bytes=1073741824
num.recovery.threads.per.data.dir=1
num.partitions=1
zookeeper.connection.timeout.ms=6000
zookeeper.connect=zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181

Will go through the ones that is essential to know.

  • offsets.topic.replication.factor: how the internal offset topic gets replicated – replication factor
  • transaction.state.log.replication.factor: how the internal transaction topic gets replicated – replication factor
  • transaction.state.log.min.isr: the minimum in sync replicas for the internal transaction topic
  • delete.topic.enable: if not true Kafka will ignore the delete topic command
  • socket.request.max.bytes: the maximum size of requests
  • log.retention.check.interval.ms: the interval to evaluate if a log should be deleted
  • log.retention.hours: how many hours a log is retained before getting deleted
  • broker.id: what is the broker id of that installation
  • log.dirs: the directories where Kafka will store the log data, can be a comma separated
  • auto.create.topics.enable: create topics if they don’t exist on sending/consuming messages or asking for topic metadata
  • num.network.threads: threads on receiving requests and sending responses from the network
  • socket.receive.buffer.bytes: buffer of the server socket
  • log.segment.bytes: the size of a log file
  • num.recovery.threads.per.data.dir: threads used for log recovery at startup and flushing at shutdown
  • num.partitions: has to do with the default number of partition a topic will have once created if partition number is not specified.
  • zookeeper.connection.timeout.ms: time needed for a client to establish connection to ZooKeeper
  • zookeeper.connect: is the list of the ZooKeeper servers

Now it’s time to create the properties for each broker. Due to the broker.id property we need to have different files with the corresponding broker.id

So our first’s brokers file would look like this (broker.id 1). Keep in mind that those brokers will run on the same docker-compose file. Therefore the zookeeper.connect property contains the internal docker compose dns names. The name of the file would be named server1.properties.

socket.send.buffer.bytes=102400
delete.topic.enable=true
socket.request.max.bytes=104857600
log.cleaner.enable=true
log.retention.check.interval.ms=300000
log.retention.hours=168
num.io.threads=8
broker.id=1
transaction.state.log.replication.factor=1
log4j.opts=-Dlog4j.configuration\=file\:/etc/kafka/log4j.properties
group.initial.rebalance.delay.ms=0
log.dirs=/var/lib/kafka
auto.create.topics.enable=true
offsets.topic.replication.factor=1
num.network.threads=3
socket.receive.buffer.bytes=102400
log.segment.bytes=1073741824
num.recovery.threads.per.data.dir=1
num.partitions=1
transaction.state.log.min.isr=1
zookeeper.connection.timeout.ms=6000
zookeeper.connect=zookeeper-1\:2181,zookeeper-2\:2181,zookeeper-3\:2181

The same recipe applies for the broker.id=2 as well as broker.id=3

After creating those three broker configuration files it is time to change our docker-compose configuration.

version: "3.8"
services:
  zookeeper-1:
    container_name: zookeeper-1
    image: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOO_MY_ID: "1"
      ZOO_SERVERS: server.1=0.0.0.0:2888:3888;2181 server.2=zookeeper-2:2888:3888;2181 server.3=zookeeper-3:2888:3888;2181
  zookeeper-2:
    container_name: zookeeper-2
    image: zookeeper
    ports:
      - "2182:2181"
    environment:
      ZOO_MY_ID: "2"
      ZOO_SERVERS: server.1=zookeeper-1:2888:3888;2181 server.2=0.0.0.0:2888:3888;2181 server.3=zookeeper-3:2888:3888;2181
  zookeeper-3:
    container_name: zookeeper-3
    image: zookeeper
    ports:
      - "2183:2181"
    environment:
      ZOO_MY_ID: "3"
      ZOO_SERVERS: server.1=zookeeper-1:2888:3888;2181 server.2=0.0.0.0:2888:3888;2181 server.3=zookeeper-3:2888:3888;2181
  kafka-1:
    container_name: kafka-1
    image: confluent/kafka
    ports:
    - "9092:9092"
    volumes:
    - type: bind
      source: ./server1.properties
      target: /etc/kafka/server.properties
  kafka-2:
    container_name: kafka-2
    image: confluent/kafka
    ports:
      - "9093:9092"
    volumes:
      - type: bind
        source: ./server2.properties
        target: /etc/kafka/server.properties
  kafka-3:
    container_name: kafka-3
    image: confluent/kafka
    ports:
      - "9094:9092"
    volumes:
      - type: bind
        source: ./server3.properties
        target: /etc/kafka/server.properties

You can find more on Compose on the Developers Essential Guide to Docker Compose.

Let’s spin up the docker-compose file.

> docker-compose -f docker-compose.yaml up

Just like the previous examples we shall run some commands through the containers.

Now that we have a proper cluster with Zookeeper and multiple Kafka brokers it is time to test them working together.
The first action is to create a topic with a replication factor of 3. The expected outcome would be for this topic to be replicated 3 kafka brokers

> docker exec -it kafka-1 /bin/bash
confluent@92a6d381d0db:/$ kafka-topics --zookeeper zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181 --create --topic tutorial-topic --replication-factor 3 --partitions 1

Our topic has been created let’s check the description of the topic.

confluent@92a6d381d0db:/$ kafka-topics --describe --topic tutorial-topic --zookeeper zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
Topic:tutorial-topic	PartitionCount:1	ReplicationFactor:3	Configs:
	Topic: tutorial-topic	Partition: 0	Leader: 2	Replicas: 2,1,3	Isr: 2,1,3

As we see the Leader for the partition is broker 2

Next step is putting some data to the topic recently created. Before doing so I will add a consumer listening for messages to that topic. While we post messages to the topic those will be printed by this consumer.

> docker exec -it kafka-3 /bin/bash
confluent@4042774f8802:/$ kafka-console-consumer --topic tutorial-topic --from-beginning --zookeeper zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181

Let’s add some topic data.

> docker exec -it kafka-1 /bin/bash
confluent@92a6d381d0db:/$ kafka-console-producer --topic tutorial-topic --broker-list kafka-1:9092,kafka-2:9092
test1
test2
test3

As expected the consumer on the other terminal will print the messages expected.

test1
test2
test3

Due to having a cluster it would be nice to stop the leader broker and see another broker to take the leadership. While doing this the expected results will be to have all the messages replicated and no disruption on consuming and publishing the messages.

Stop the leader which is broker-2

> docker stop kafka-2

Check the leadership from another broker

confluent@92a6d381d0db:/$ kafka-topics --describe --topic tutorial-topic --zookeeper zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
Topic:tutorial-topic	PartitionCount:1	ReplicationFactor:3	Configs:
	Topic: tutorial-topic	Partition: 0	Leader: 1	Replicas: 2,1,3	Isr: 1,3

The leader now is kafka-1

Read the messages to see that they did got replicated.

> docker exec -it kafka-3 /bin/bash
confluent@4042774f8802:/$ kafka-console-consumer --topic tutorial-topic --from-beginning --zookeeper zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
test1
test2
test3

As expected apart from the Leadership being in place our data have also been replicated!

If we try to post new messages, it will also be a successful action.

So to summarise we did run a Kafka cluster connected to a zookeeper ensemble. We did create a topic with replication enabled to 3 brokers and last but not least we did test what happens if one broker goes down.

On the next blog we are going to wrap it up so our local machine clients can connect to the docker compose ensemble.

Kafka & Zookeeper for Development: Zookeeper Ensemble

Previously we spun up Zookeeper and Kafka locally but also through Docker. What comes next is spinning up more than just one Kafka and Zookeeper node and create a 3 node cluster. To achieve this the easy way locally docker-compose will be used. Instead of spinning up various instances on the cloud or running various Java processes and altering configs, docker-compose will greatly help us to bootstrap a Zookeeper ensemble and the Kafka brokers, with everything needed preconfigured.

The first step is to create the Zookeeper ensemble but before going there let’s check the ports needed.
Zookeeper needs three ports.

  • 2181 is the client port. On the previous example it was the port our clients used to communicate with the server.
  • 2888 is the peer port. This is the port that zookeeper nodes use to talk to each other.
  • 3888 is the leader port. The port that nodes use to talk to each other when it comes to the leader election.

By using docker compose our nodes will use the same network and the container name will also be an internal dns entry.
The names on the zookeeper nodes would be zookeeper-1, zookeeper-2, zookeeper-3.

Our next goal is to give to each zookeeper node a configuration that will enable the nodes to discover each other.

This is the typical zookeeper configuration expected.

  • tickTime is the unit of time in milliseconds zookeeper uses for heartbeat and minimum session timeout.
  • dataDir is the location where ZooKeeper will store the in-memory database snapshots
  • initlimit and SyncLimit are used for the zookeeper synchronization.
  • server* are a list of the nodes that will have to communicate with each other

zookeeper.properties

clientPort=2181
dataDir=/var/lib/zookeeper
syncLimit=2
DATA.DIR=/var/log/zookeeper
initLimit=5
tickTime=2000
server.1=zookeeper-1:2888:3888
server.2=zookeeper-2:2888:3888
server.3=zookeeper-3:2888:3888

When it comes to a node the server that the node is located, should be bound to `0.0.0.0`. Thus we need three different zookeeper properties per node.

For example for the node with id 1 the file should be the following
zookeeper1.properties

clientPort=2181
dataDir=/var/lib/zookeeper
syncLimit=2
DATA.DIR=/var/log/zookeeper
initLimit=5
tickTime=2000
server.1=0.0.0.0:2888:3888
server.2=zookeeper-2:2888:3888
server.3=zookeeper-3:2888:3888

The next question that arises is the id file of zookeeper. How a zookeeper instance can identify which is its id.

Based on the documentation we need to specify the server ids using the myid file

The myid file is a plain text file located at a nodes dataDir containing only a number the server name.

So three myids files will be created each containing the number of the broker

myid_1.txt

1

myid_2.txt

2

myid_3.txt

3

It’s time to spin up the Zookeeper ensemble. We will use the files specified above. Different client ports are mapped to the host to avoid collision.

version: "3.8"
services:
  zookeeper-1:
    container_name: zookeeper-1
    image: zookeeper
    ports:
      - "2181:2181"
    volumes:
      - type: bind
        source: ./zookeeper1.properties
        target: /conf/zoo.cfg
      - type: bind
        source: ./myid_1.txt
        target: /data/myid
  zookeeper-2:
    container_name: zookeeper-2
    image: zookeeper
    ports:
      - "2182:2181"
    volumes:
    - type: bind
      source: ./zookeeper2.properties
      target: /conf/zoo.cfg
    - type: bind
      source: ./myid_2.txt
      target: /data/myid
  zookeeper-3:
    container_name: zookeeper-3
    image: zookeeper
    ports:
      - "2183:2181"
    volumes:
      - type: bind
        source: ./zookeeper3.properties
        target: /conf/zoo.cfg
      - type: bind
        source: ./myid_3.txt
        target: /data/myid

Eventually instead of having all those files mounted it would be better if we had a more simple option. Hopefully the image that we use gives us the choice of specifying ids and brokers using environment variables.

version: "3.8"
services:
  zookeeper-1:
    container_name: zookeeper-1
    image: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOO_MY_ID: "1"
      ZOO_SERVERS: server.1=0.0.0.0:2888:3888;2181 server.2=zookeeper-2:2888:3888;2181 server.3=zookeeper-3:2888:3888;2181
  zookeeper-2:
    container_name: zookeeper-2
    image: zookeeper
    ports:
      - "2182:2181"
    environment:
      ZOO_MY_ID: "2"
      ZOO_SERVERS: server.1=zookeeper-1:2888:3888;2181 server.2=0.0.0.0:2888:3888;2181 server.3=zookeeper-3:2888:3888;2181
  zookeeper-3:
    container_name: zookeeper-3
    image: zookeeper
    ports:
      - "2183:2181"
    environment:
      ZOO_MY_ID: "3"
      ZOO_SERVERS: server.1=zookeeper-1:2888:3888;2181 server.2=0.0.0.0:2888:3888;2181 server.3=zookeeper-3:2888:3888;2181

You can find more on Compose on the Developers Essential Guide to Docker Compose.

Now let’s check the status of our zookeeper nodes.

Let’s use the zookeeper shell to see the leaders and followers

docker exec -it zookeeper-1 /bin/bash
>./bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: follower

And

> docker exec -it zookeeper-3 /bin/bash
root@81c2dc476127:/apache-zookeeper-3.6.2-bin# ./bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: leader

So in this tutorial we created a Zookeeper ensemble using docker-compose and we also had a leader election. This recipe can be adapted to apply on the a set of VMs or a container orchestration engine.

On the next tutorial we shall add some Kafka brokers to connect to the ensemble.

Kafka & Zookeeper for Development: Local and Docker

Kafka popularity increases every day more and more as it takes over the streaming world. It is already provided out of the box on cloud providers like AWS, Azure and IBM Cloud.
Eventually for cases of local development it is a bit peculiar due to requiring various moving parts.

This blog will focus on making it easy for a developer to spin up some Kafka instances on a local machine without having to spin up VMs on the cloud.

We shall start with the usual Zookeeper and Kafka configuration. The example bellow will fetch a specific version so after some time is good to check the Apache Website.

> wget https://www.mirrorservice.org/sites/ftp.apache.org/kafka/2.6.0/kafka_2.13-2.6.0.tgz
> tar xvf kafka_2.13-2.6.0.tgz
> cd kafka_2.13-2.6.0

We just downloaded Kafka locally and now is the time to Spin up Kafka.

First we should spin up the Zookeeper

> ./bin/zookeeper-server-start.sh config/zookeeper.properties

Then spin up the Kafka instance

> ./bin/kafka-server-start.sh config/server.properties

As you see we only spun up one instance of Kafka & Zookeeper. This is way different from what we do in production where ZooKeeper servers should be deployed on multiple nodes. More specific 2n + 1 ZooKeeper servers where n > 0 need to be deployed. This number helps the ZooKeeper ensemble to perform majority elections for leadership.

In our case for local development one Kafka broker and one Zookeeper instances are enough in order to create and consume a topic.

Let’s push some messages to a topic. There is no need to create the topic, pushing a message will create it.

bin/kafka-console-producer.sh --topic tutorial-topic --bootstrap-server localhost:9092
>a
>b
>c

Then let’s read it. Pay attention to the –from-beginning flag, all the messages submitted from the beginning shall be read.

bin/kafka-console-consumer.sh --topic tutorial-topic --from-beginning --bootstrap-server localhost:9092
>a
>b
>c

Now let’s try and do this using docker. The advantage of docker is that we can run Kafka on a local docker network and add as many machines as needed and establish a Zookeeper ensemble the easy way.

Start zookeeper first

docker run --rm --name zookeeper -p 2181:2181 confluent/zookeeper

And then start your docker container after doing a link with the zookeeper container.

docker run --rm --name kafka -p 9092:9092 --link zookeeper:zookeeper confluent/kafka

Let’s create the messages through docker. As with most docker images you already have the tools needed bundled inside the image.
So the publish command would very close to the command we executed previously.

> docker exec -it kafka /bin/bash
kafka-console-producer --topic tutorial-topic --broker-list localhost:9092
a
b
c

The same applies for the consume command.

> docker exec -it kafka /bin/bash
kafka-console-consumer --topic tutorial-topic --from-beginning --zookeeper zookeeper:2181
a
b
c

That’s it! We Just run Kafka locally for local development seamlessly!

Testing with Hoverfly and Java Part 3: State

Previously we simulated a delay scenario using Hoverfly. Now it’s time to dive deeper and go for a state based testing. By doing a stateful simulation we can change the way the tests endpoints behave based on how the state changed.

 

Hoverfly does have a state capability. State in a hoverfly simulation is like a map. Initially it is empty but you can define how it will get populated per request.

Our strategy would be to have a request that initializes the state and then specifies other requests that change that state.

public class SimulationStateTests {

	private Hoverfly hoverfly;

	@BeforeEach
	void setUp() {
		var simulation = SimulationSource.dsl(service("http://localhost:8085")
				.get("/initialize")
				.willReturn(success("{\"initialized\":true}", "application/json")
						.andSetState("shouldSucceed", "true")
				)
				.get("/state")
				.withState("shouldSucceed", "false")
				.willReturn(serverError().andSetState("shouldSucceed", "true"))
				.get("/state")
				.withState("shouldSucceed", "true")
				.willReturn(success("{\"username\":\"test-user\"}", "application/json")
						.andSetState("shouldSucceed", "false"))

		);

		var localConfig = HoverflyConfig.localConfigs().disableTlsVerification().asWebServer().proxyPort(8085);
		hoverfly = new Hoverfly(localConfig, SIMULATE);
		hoverfly.start();
		hoverfly.simulate(simulation);
	}

	@AfterEach
	void tearDown() {
		hoverfly.close();
	}

}

Unfortunately on the state we can only specify values in a key value fashion and not by passing a function for a key.
However with the right workaround many scenarios could be simulated.

In the example we first initialize the state and the we issue requests that behave differently based on the state, but also they do change the state.

So we expect to have a continuous first succeed and then fail mode, which can be depicted in the following test.

	@Test
	void testWithState() {
		var client = HttpClient.newHttpClient();
		var initializationRequest = HttpRequest.newBuilder()
				.uri(URI.create("http://localhost:8085/initialize"))
				.build();
		var initializationResponse = client.sendAsync(initializationRequest, HttpResponse.BodyHandlers.ofString())
				.thenApply(HttpResponse::body)
				.join();
		Assertions.assertEquals("{\"initialized\":true}", initializationResponse);

		var statefulRequest = HttpRequest.newBuilder()
				.uri(URI.create("http://localhost:8085/state"))
				.build();

		for (int i = 0; i < 100; i++) {
			var response = client.sendAsync(statefulRequest, HttpResponse.BodyHandlers.ofString())
					.join();

			int statusCode = i % 2 == 0 ? 200 : 500;

			Assertions.assertEquals(statusCode, response.statusCode());
		}
	}

That’s all about stateful simulation. On the next part we shall proceed on Hoverfly matchers