Forem: Marcos Maia

Spring Kafka Streams playground with Kotlin - V

Marcos Maia — Sun, 13 Feb 2022 11:11:56 +0000

This post is the final part of a series where we create a simple Kafka Streams Application with Kotlin using Spring boot and Spring Kafka.

Please check the first part of the tutorial to get started and get further context of what we're building.

If you want to start from here you can clone the source code for this project git clone git@github.com:mmaia/simple-spring-kafka-stream-kotlin.git and then checkout v8 git checkout v8 and follow from there continuing with this post.

In this post we're going to use the QuoteStream we created in the last post and use grouping by key to aggregate and count the total volume(count) per stock quote symbol and also materialize it with windowing so we can create an endpoint and query the local WindowStore to calculate the volume for each instrument in specific intervals. We will create an endpoint on our Controller to enable us to expose that information. Have fun.

Grouping and counting

In this part of the series we will use Kafka Streams to keep track of the number of quotes for each key and then expand that to be able to check the number of quotes for a specific interval in time, in the finance lingo where a new Quote represent an executed trade this counting is also known as Trade Volume.

For this part we will use two new topics to store the total count of quotes per symbol and another one to store the total count per interval, for that let's start adding the following to the appTopics function on the KafkaConfiguration class so these topics are created for use next time we start the application:


 TopicBuilder
.name(COUNT_TOTAL_QUOTES_BY_SYMBOL_TOPIC)
.compact()
.build(),

TopicBuilder
.name(COUNT_WINDOW_QUOTES_BY_SYMBOL_TOPIC)
.build()

And let's add some configuration we will need to the bottom of KafkaConfiguration class and also move the serdeCofnig from the LeverageStream to the KafkaConfiguration global section so we can re-use it in our QuoteStream next:

const val COUNT_TOTAL_QUOTES_BY_SYMBOL_TOPIC = "count-total-by-symbol-topic"
const val COUNT_WINDOW_QUOTES_BY_SYMBOL_TOPIC = "count-window-by-symbol-topic"
const val QUOTES_BY_WINDOW_TABLE = "quotes-by-window-table"
val serdeConfig: MutableMap<String, String> = Collections.singletonMap(
   AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, SCHEMA_REGISTRY_URL
)

Now we declare a new Avro type which we will use to store the quotes per window we will be counting, create a new file called quote-per-window.avsc and add the following content to it:

{
  "namespace": "com.maia.springkafkastream.repository",
  "type": "record",
  "name": "QuotesPerWindow",
  "fields": [
    { "name": "symbol", "type": "string"},
    { "name": "startTime", "type": "long"},
    { "name": "endTime", "type":  "long"},
    { "name": "count", "type":  "long"}
  ]
}

Build your project so the new java type is generated: mvn clean package -DskipTests

Grouping and counting

We start by doing a simple total count per symbol and send that to the newly created topic count-total-by-symbol-topic, so just after the segment where we did the branching in the last post add the following lines:

// group by key
val groupedBySymbol: KGroupedStream<String, ProcessedQuote> = resStream.groupByKey()

// count and send to compact topic so we can always quickly access the total count by symbol
val quotesCount: KTable<String, Long> = groupedBySymbol.count()
quotesCount.toStream().to(COUNT_TOTAL_QUOTES_BY_SYMBOL_TOPIC, Produced.with(Serdes.String(), Serdes.Long()))

That's it, quite simple, this will keep track and count every single different quote by it's key and keep track of total count, to test you can run the app and consume messages from the count-total-by-symbol-topic and after you send some quotes using the API we created in a previous post you can see the updates on the topic:

But to be honest, in this case having the total count doesn't seem to useful, it would be more useful if we could calculate the Trade Volume in specific interval, yeah cool, let's do that!

Create a KTable and materialize it so we can then use an endpoint and query the trade volume per symbol per interval, in the same QuoteStream.quoteKStream method add the following code:

// we could transform it materialize it to be able to query directly from kafka
val quotesWindowedKTable: KTable<Windowed<String>, Long> = groupedBySymbol
   .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(30)))
    .count(
        Materialized.`as`<String, Long, WindowStore<Bytes, ByteArray>>(QUOTES_BY_WINDOW_TABLE)
            .withValueSerde(Serdes.Long())
    )

So this piece of code above is storing the count per 30 second interval for each key using a WindowStore which we can then query later to calculate volumes per interval.

Please make sure you understand the difference and consequence of using KTable vs GlobalKTables specially if you have multiple instances of your client application running.

To be able to query the created WindowStore we need to declare a ReadOnlyWindowStore and initialize it, declare it as an instance variable:

private lateinit var quotesPerWindowView: ReadOnlyWindowStore<String, Long>

And then declare a listener to initialize it:

@Bean
fun afterStartQuote(sbfb: StreamsBuilderFactoryBean): StreamsBuilderFactoryBean.Listener {
    val listener: StreamsBuilderFactoryBean.Listener = object : StreamsBuilderFactoryBean.Listener {
        override fun streamsAdded(id: String, streams: KafkaStreams) {
            quotesPerWindowView = streams
                .store(
                    StoreQueryParameters
                        .fromNameAndType(QUOTES_BY_WINDOW_TABLE, QueryableStoreTypes.windowStore())
                )
        }
    }
    sbfb.addListener(listener)
    return listener
}

Sweet, now that we have an initialized object for this WindowStore let's create a function that calculates the volume based on the quote symbol, start and end time specified:

fun quoteVolumePerInterval(key: String, start: Long, end: Long): QuotesPerWindow {
    val listOfCountedQuotes: WindowStoreIterator<Long> =
        quotesPerWindowView.fetch(key, Instant.ofEpochMilli(start), Instant.ofEpochMilli(end))

    var totalCountForInterval = 0L
    listOfCountedQuotes.forEach {
        totalCountForInterval += it.value
    }

    return QuotesPerWindow(key, start, end, totalCountForInterval)
}

We can then create a new REST endpoint to expose the volume, to simplify we will create one where the callers specifies the minutes from now and we calculate the volume of the past specified volumes for that specific instrument. Add the following function to the QuotesController class:

@GetMapping("/quotes/count/{symbol}")
fun getQuotesPerWindow(
    @PathVariable symbol: String,
    @RequestParam("pastMinutes") pastMinutes: Int
): ResponseEntity<QuotesPerWindowDTO> {
    val end = Instant.now()
    val start = end.minusSeconds(pastMinutes.toLong() * 60)
    val quotesPerWindow: QuotesPerWindow = quoteStream.quoteVolumePerInterval(symbol, start.toEpochMilli(), end.toEpochMilli())
    val result = QuotesPerWindowDTO(symbol, quotesPerWindow.count, start, end)
    return ResponseEntity.ok(result)
}

Build and run the application and let's play around with it a bit. Create some quotes as before send them to Kafka and with a new client call like the one before check the volume for those quotes in the specified interval:

GET http://localhost:8080/api/quotes/count/APPL?pastMinutes=5

That's really nice, but let's now consider that you might want to calculate those totals and then be able to send them to Elasticsearch or MongoDB for further processing, the current KTable doesn't have much context as it's built in Kafka elements like the Windowed store, what you want to do in this case is to send this data to a new topic with the total count, the interval and the symbol so you can easily plugin a Kafka Connector to Synchronize those count aggregations to your external storage, let's do that, add the following code to the QuoteStream.quoteKStream function just after the KTable definition we did above:

// and then transform and send it to a new topic which we could then use a connector and send to elastic or mongodb for example...
quotesWindowedKTable
    .toStream()
    .map(processedQuoteToQuotesPerWindowMapper::apply)
    .to(COUNT_WINDOW_QUOTES_BY_SYMBOL_TOPIC, Produced.with(Serdes.String(), quotesPerWindowSerde))

You will get an error at this point due to the missing definition of the processedQuoteToQuotesPerWindowMapper which is a Bifunction which receives as input the the value type of our KTable Windowed<String>, Long and we than transform it to the Avro type we created earlier so we can send this to a Topic with meaningful context, so let's add a property function to represent that transformation and fix the error, just after the quotesPerWindowView declaration at the class level add:

/**
 * From processed quote to quotes per window
 */
var processedQuoteToQuotesPerWindowMapper =
    BiFunction<Windowed<String>, Long, KeyValue<String, QuotesPerWindow>> { key, value ->
        val quotesPerWindow = QuotesPerWindow(key.key(), key.window().start(), key.window().end(), value ?: 0)
        KeyValue(key.key(), quotesPerWindow)
    }

And we also need to initialize the Serde we used, so add a new @PostConstruct function to take care of that:

@PostConstruct
fun init() {
    quotesPerWindowSerde.configure(serdeConfig, false)
}

As usual you can checkout the project to this point using git checkout v9 or just checkout latest main git checkout main, have fun.

That's it it's done, you can now play around with it, check the content of the new topic and you have a nice Kafka Streams playground to expand and practice.

Thank you so much to have reached this far, this is the end of a very long series of articles, took me a lot of effort to do it, I really appreciate feedback with issues and typos you will most likely find so I can improve it. Constructive suggestions are also Welcome!

Photo by Kevin Ku on Unsplash

Spring Kafka Streams playground with Kotlin - IV

Marcos Maia — Sat, 12 Feb 2022 16:04:45 +0000

Context

This post is part of a series where we create a simple Kafka Streams Application with Kotlin using Spring boot and Spring Kafka.

Please check the first part of the tutorial to get started and get further context of what we're building.

If you want to start from here you can clone the source code for this project git clone git@github.com:mmaia/simple-spring-kafka-stream-kotlin.git and then checkout v7 git checkout v7 and follow from there continuing with this post.

In this post we're going to create our QuoteStream, we will process messages from the quote topic and do a join with the GlobalKTable for leverage we created in the previous post, we will then do branching and send quotes to three different topics based on it's keys.

Creating a Quote Stream

Let's now create a stream from the stock-quotes-topic and join with Leverage GlobalKTable we created in the last post, we will use a stream join and if there's a Leverage for that specific quote we enrich the quote and publish it to one topic based on it's key. This will demonstrate the usage of branching to separate data for post processing which is quite common use case.

In our sample application we will produce the data to one of three new topics, we will produce ant Apple quotes to a topic, Google quotes to another topic and all other quotes to the third topic.

AAPL -> apple-stocks-topic
GOOGL -> google-stocks-topic
Any others -> all-other-stocks-topic

In order to do that our first step is to create those topics using the Admin client. Add the following constants to the KafkaCofiguration class:

const val AAPL_STOCKS_TOPIC = "apple-stocks-topic"
const val GOOGL_STOCKS_TOPIC = "google-stocks-topic"
const val ALL_OTHER_STOCKS_TOPIC = "all-other-stocks-topic"

Change the appTopics function on the same class to also create those three new topics:

@Bean
fun appTopics(): NewTopics {
    return NewTopics(
        TopicBuilder.name(STOCK_QUOTES_TOPIC).build(),
        TopicBuilder.name(LEVERAGE_PRICES_TOPIC)
            .compact().build(),
        TopicBuilder.name(AAPL_STOCKS_TOPIC).build(),
        TopicBuilder.name(GOOGL_STOCKS_TOPIC).build(),
        TopicBuilder.name(ALL_OTHER_STOCKS_TOPIC).build(),
    )
}

The next time you run the application those new topics will be created using the admin client.

Let's now add the new schema definition which we will use for those new topics, create a new avro-schema file under src > main > avro called processed-quote with the following content:

{
  "namespace": "com.maia.springkafkastreamkotlin.repository",
  "type": "record",
  "name": "ProcessedQuote",
  "fields": [
    { "name": "symbol", "type": "string"},
    { "name": "tradeValue", "type": "double"},
    { "name": "tradeTime", "type": ["null", "long"], "default": null},
    { "name": "leverage", "type": ["null", "double"], "default":  null}
  ]
}

Notice that the difference in this case is just a new leverage field which we will use to enrich the incoming quote with the value if they match. Build the project so the java code is generate for this new avro schema.

mvn clean package -DskipTests

Let's now create a new class on the repository package called QuoteStream, we will need a reference to our Leverage GlobalKTable so we use Spring Boot dependency injection:

@Repository
class QuoteStream(val leveragepriceGKTable: GlobalKTable<String, LeveragePrice>) {
...

On this class declare a function to process and enrich the quote:

@Bean
fun quoteKStream(streamsBuilder: StreamsBuilder): KStream<String, ProcessedQuote> {

In this function create a KStream which will process the data from the stock-quotes-topic, do a join with the GlobalKTable we created for leverage and transform in the new Avro Type ProcessedQuote enriching the quotes with leverage if it's available:

val stream: KStream<String, StockQuote> = streamsBuilder.stream(STOCK_QUOTES_TOPIC)

val resStream: KStream<String, ProcessedQuote> = stream
    .leftJoin(leveragepriceGKTable,
        { symbol, _ -> symbol },
        { stockQuote, leveragePrice ->
            ProcessedQuote(
                stockQuote.symbol,
                stockQuote.tradeValue,
                stockQuote.tradeTime,
                leveragePrice?.leverage
            )
        }
    )

and to wrap up this function we will tap in the new Stream and based on the Key we will send the message to specific topics and return the new stream so we can re-use it later for other operations:

KafkaStreamBrancher<String, ProcessedQuote>()
            .branch({ symbolKey, _ -> symbolKey.equals("APPL", ignoreCase = true) }, { ks -> ks.to(AAPL_STOCKS_TOPIC) })
            .branch({ symbolKey, _ -> symbolKey.equals("GOOGL",ignoreCase = true) }, { ks -> ks.to(GOOGL_STOCKS_TOPIC) })
            .defaultBranch { ks -> ks.to(ALL_OTHER_STOCKS_TOPIC) }
            .onTopOf(resStream)

    return resStream
}

If you just want to get your local code to this point without using the presented code you can checkout v8: git checkout v8

Cool, now we can play around a bit with it, let's build and run our application(make sure your local kafka setup is running):

mvn clean package -DskipTests && mvn spring-boot:run

You can then send some leverage messages and quote messages using the APIs as we did before in this tutorial.

And you can check the messages being enriched if the specific quote has a leverage and flowing to the three different topics based on their keys, here's some screenshots from the topics I took while playing around with the project using Conduktor.

The messages seem duplicated on the screenshots but that's because I sent them multiple times with the same values while playing around. I also sent a few before sending the respective leverage so you can see what happens and check that the initial ones on the bottom have a null leverage.

That's it for now. Tomorrow I will be publishing part V of this tutorial where we will use grouping, counting and calculate volume using Kafka Streams DSL. Stay tuned.

Photo by Nubelson Fernandes on Unsplash

Spring Kafka Streams playground with Kotlin - III

Marcos Maia — Fri, 11 Feb 2022 15:38:53 +0000

Context

This post is part of a series where we create a simple Kafka Streams Application with Kotlin using Spring boot and Spring Kafka.

Please check the first part of the tutorial to get started and get further context of what we're building.

If you want to start from here you can clone the source code for this project git clone git@github.com:mmaia/simple-spring-kafka-stream-kotlin.git and then checkout v6 git checkout v6 and follow from there continuing with this post.

In this post we're going to create our first Spring Kafka Streams, GlobalKTable with a RocksDB state store and expose it on our Rest Controller created in part II using a ReadOnlyKeyValueStore implementation.

Configure App for Spring Kafka Streams

Now that we're going to use Kafka Streams we need to let Spring Boot know that so we start adding the @EnableKafkaStreams Spring Kafka annotation to the kafkaConfiguration class definition, just below the @EnableKafka one we already have:

@Configuration
@EnableKafka
@EnableKafkaStreams
class KafkaConfiguration { 
...

We also need to add the basic default streams configuration the same way we did to the consumers and producers before but this time we will use the programmatic approach instead of adding as a Spring Boot configuration to the application.yaml file, which also would work but it's nice to have examples of both approaches so you're aware of them and can pick the one that suits you the best.

In case you have multiple consumers, producers or streams in the same Spring Boot application and want different configurations for each you will have to take the programmatic approach as it gives you more flexibility to configure each specific client while the configuration approach only enables you to configure the default spring bean of each type.

In the KafkaConfiguration class we create the default Streams configuration:

@Bean(name = [KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME])
    fun defaultKafkaStreamsConfig(): KafkaStreamsConfiguration {
        val props: MutableMap<String, Any> = HashMap()
        props[StreamsConfig.BOOTSTRAP_SERVERS_CONFIG] = KAFKA_HOSTS
        props[StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG] =
            LogAndContinueExceptionHandler::class.java
        props[AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG] = SCHEMA_REGISTRY_URL
        props[StreamsConfig.APPLICATION_ID_CONFIG] = "quote-stream"
        props[StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG] = Serdes.String().javaClass.name
        props[StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG] = SpecificAvroSerde::class.java
        props[ConsumerConfig.GROUP_ID_CONFIG] = "stock-quotes-stream-group"

        return KafkaStreamsConfiguration(props)
    }

and we will also add to the KafkaConfiguration class another bean so we can print out the Stream states which is good for learning and understanding purposes, in a real application you could choose to not do this and, if you do, you would use a logger instead of stdout printing it like we do in this example.

@Bean
fun configurer(): StreamsBuilderFactoryBeanConfigurer? {
    return StreamsBuilderFactoryBeanConfigurer { fb: StreamsBuilderFactoryBean ->
        fb.setStateListener { newState: KafkaStreams.State, oldState: KafkaStreams.State ->
            println("State transition from $oldState to $newState")
        }
    }
}

Create a GlobalKTable for Leverage

Now that our Kafka Streams are properly configured we can create our first Kafka Stream element. We will start creating a GlobalKTable and a materialized view to the Leverage prices.

Create a new class called LeverageStream in the same repository package, annotate it with @Repository.

@Repository
class LeverageStream {
...

Now before creating our GlobalKTable and Materialize it we need to tell Kafka Streams which type of Serialization and Deserialization it will need for the object we will process and store. In Kafka Streams this is done using a Serde and because we are using Avro we will use an Avro Serde SpecificAvroSerde passing in our LeveragePrice type we have created in a previous step, add the following to the LeverageStream class to declare the Serde, configure and initialize it.

private val leveragePriceSerde = SpecificAvroSerde<LeveragePrice>()

val serdeConfig: MutableMap<String, String> = Collections.singletonMap(       AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, SCHEMA_REGISTRY_URL)

@PostConstruct
fun init() {
    leveragePriceSerde.configure(serdeConfig, false)
}

Now that we told Kafka Streams the Serde and the configuration where to get the Avro Schema we can add the following code which creates a GlobalKTable with a materialized RocksDB KeyValue store.

    @Bean
    fun leveragePriceBySymbolGKTable(streamsBuilder: StreamsBuilder): GlobalKTable<String, LeveragePrice> {      
        return streamsBuilder
            .globalTable(
                LEVERAGE_PRICES_TOPIC,
                Materialized.`as`<String, LeveragePrice, KeyValueStore<Bytes, ByteArray>>(LEVERAGE_BY_SYMBOL_TABLE)
                    .withKeySerde(Serdes.String())
                    .withValueSerde(leveragePriceSerde)
            )
    }

The materialized view will by default create a RocksDB locally to your application for you, you don't have to manage or explicitly do anything it's... Magic!

You might be wondering where the hack is this RocksDB being created, well you can control that using a configuration StreamsConfig.STATE_DIR_CONFIG i.e - state.dir and point it to the location you want. In this case I left the defaults as we didn't specify the property so it will pick the default depending on the O.S you're running this tutorial. On linux machines you will find it under /tmp/kafka-streams/${stream-id}/${type}/rocksdb/${local_storage_name} folder, the full path for this case is: /tmp/kafka-streams/quote-stream/global/rocksdb/leverage-by-symbol-ktable/ where you can see the RocksDB files:

Expose Leverage Materialized view

Great, now that we have a local storage we want to be able to query it to get the most up to date Leverage for the instruments.

In order to be able to query RocksDB we will declare a ReadOnlyKeyValueStore, we will use Kotlin's lateinit modifier as we need to bind in the Stream lifecycle before we initialize it. i.e - the RocksDB storage needs to be in place. Declare it in LeverageStream:

private lateinit var leveragePriceView: ReadOnlyKeyValueStore<String, LeveragePrice>

and then declare a @Bean listener to initialized it once the Stream is created:

@Bean
fun afterStart(sbfb: StreamsBuilderFactoryBean): StreamsBuilderFactoryBean.Listener {
    val listener: StreamsBuilderFactoryBean.Listener = object : StreamsBuilderFactoryBean.Listener {
        override fun streamsAdded(id: String, streams: KafkaStreams) {
            leveragePriceView = streams.store<ReadOnlyKeyValueStore<String, LeveragePrice>>(
                    StoreQueryParameters.fromNameAndType(
                        LEVERAGE_BY_SYMBOL_TABLE,
                        QueryableStoreTypes.keyValueStore()
                    )
            )
        }
    }
    sbfb.addListener(listener)
    return listener
}

We can then create a function that we use to get the specific value passing in a key:

fun getLeveragePrice(key: String?): LeveragePrice? {
    return leveragePriceView[key]
}

Finally we can then add a new REST endpoint to our QuotesController class to expose the Leverage:

    @GetMapping("leveragePrice/{instrumentSymbol}")
    fun getLeveragePrice(@PathVariable instrumentSymbol: String): ResponseEntity<LeveragePriceDTO> {
        val leveragePrice: LeveragePrice = leverageStream.getLeveragePrice(instrumentSymbol)
            ?: return ResponseEntity.noContent().build()
        // if quote doesn't exist in our local store we return no content.
        val result = LeveragePriceDTO(leveragePrice.symbol.toString(), BigDecimal.valueOf(leveragePrice.leverage))
        return ResponseEntity.ok(result)
    }

With that we can build and run our application: mvn clean package -DskipTests && mvn spring-boot:run. Make sure to have a local Kafka running as explained in part 1 and part 2 of this series and execute some calls:

Add a leverage which will send a new Leverage to the Kafka Topic:

POST http://localhost:8080/api/leverage
Content-Type: application/json

< ./test-data/leveragePrice1.json

Query it from the RocksDB local storage:

GET http://localhost:8080/api/leveragePrice/APPL

Add a few more quotes and play around querying it so you'll see how the storage is automatically updated with the most up to date quote you've sent.

As usual you can check the code up to this point checking out the project repo with git clone git@github.com:mmaia/simple-spring-kafka-stream-kotlin.git and check out v7 with git checkout v7.

This is it for part 3 of this series, I hope you had some fun learning and playing with Kafka Streams so far. In the next post of this series we will create a Stream for Quotes, Join with Leverage and branch producing it to different topics based on some criteria, stay tuned.

Have fun, be kind to yourself and others!

Photo by Ryland Dean on Unsplash

Spring Kafka Streams playground with Kotlin - II

Marcos Maia — Thu, 10 Feb 2022 14:16:40 +0000

Context

This post is part of a series where we create a simple Application with Kotlin using Spring boot and Spring Kafka.

Please check the first part of the tutorial to get started and get further context of what we're building.

If you want to start from here you can clone the source code for this project git clone git@github.com:mmaia/simple-spring-kafka-stream-kotlin.git and then checkout v4 git checkout v4 and follow from there continuing with this post.

In this second part we will build the producers LeverageProducer and QuotesProducer and our Controller QuoteController and also use the Kafka client Admin to create the topics for us.

We will then be able to send some messages using the REST endpoinst we created and check the messages in Kafka using the command line client or Conduktor. Let's go for it.

Setting up Kafka Admin to create topics

In this section we will create the spring-kafka setup so Kafka Admin Client create the topics for us on the broker on Application startup.

For this sample application I will keep the package structure very simple and we will end up with two subpackages, api where our rest endpoints will be created and repository where we will put all kafka configuration and kafka client code. In real projects I usually have other layers and I tend to use business related packaging paths to segregate subdomain, but this is to be discussed in another post.

Create a package called repository and a new Kotlin class KafkaConfiguraton in this new package.

Add the Spring boot @Configuration and Spring Kafka @EnableKafka annotations to your class, create constants with the topic names and a Spring @Bean with spring-kafka Admin code to create the two initial topics for leverage and quotes. See code fragment below, notice that we will create the leverage topic as a Kafka compact topic. We will later build a Global KTable to read this data, we will also use the default settings for this simple example.

KafkaConfiguration Kotlin class:


@Configuration
@EnableKafka
class KafkaConfiguration {

    // TOPICS
    val quotesTopic = "stock-quotes-topic"
    val leveragePriceTopic = "leverage-prices-topic"

    @Bean
    fun appTopics(): NewTopics? {
        return NewTopics(
          TopicBuilder.name(quotesTopic).build(),
          TopicBuilder.name(leveragePriceTopic)
            .compact().build(),
        )
    }
}

Add the following property to the application.yaml under src/main/resources. If your file is named application.properties you can just change the name / extension to yaml instead.


spring:
  kafka:
    bootstrap-servers: localhost:9092

You may git checkout v5 to check out the code up to this point.

Start local Kafka on compose or Kubernetes and then build and run this application from your IDE or from command line:

mvn clean package -DskipTests

run it:

mvn spring-boot:run

Now when the application starts you should see the Kafka Client Admin configuration info on the logs:

Ok, once the application starts you can now check on the kafka broker that the two topics are now available due to the Admin client code we just created.

Enter the running container: kubectl exec -it kafka-0 -- bash and then list topics kafka-topics --list --bootstrap-server kafka:29092, you should see the topics:

Use exit to go back to the host command line.

Create Kafka producers

In this next section we will create the Kafka producers and API endpoints to send messages to the topics we just created.

Kafka Producers are thread safe and we could use a single producer passing in different types of Schemas and topics but in this tutorial to avoid abstractions we will make it explicit creating two producers.

First lets add the generic producer configuration to spring-kafka producers to the application.yaml file, under spring.kafka add:


    producer:
      bootstrap-servers: localhost:9092
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
      properties:
        schema.registry.url: http://localhost:8081

The configurations tell our producer clients where to find Kafka and the Schema registry and the types of the key/value pairs we will be sending to our topics.

Creating the producers:

In the producers we will need to reference the topics to send messages, it would be better to have a central point from where to get them. Just now we added the topic names to the KafkaConfiguration class but they are normal Kotlin val attributes, let's make a change and add them as constants instead so we can easily access them, change the KafkaConfiguration class to be like this so we can have access to those constants from our producers:


@Configuration
@EnableKafka
class KafkaConfiguration {
    @Bean
    fun appTopics(): NewTopics {
        return NewTopics(
            TopicBuilder.name(STOCK_QUOTES_TOPIC).build(),
            TopicBuilder.name(LEVERAGE_PRICES_TOPIC).compact().build(),
        )
    }
}

// constants for topics
const val STOCK_QUOTES_TOPIC = "stock-quotes-topic"
const val LEVERAGE_PRICES_TOPIC = "leverage-prices-topic"

Notice that above we moved the constants outside of the class definition.

Under the repository package create two new Kotlin classes called LeveragePriceProducer and QuotePriceProducer and add the following content to them:

LeveragePriceProducer class:


@Repository
class LeveragePriceProducer(val leveragePriceProducer: KafkaTemplate<String, LeveragePrice> ) {

    fun send(message: LeveragePrice) {
        leveragePriceProducer.send(LEVERAGE_PRICES_TOPIC, message.symbol.toString(), message)
    }

}

StockQuoteProducer class:


@Repository
class StockQuoteProducer(val quoteProducer: KafkaTemplate<String, StockQuote>) {

    fun send(message: StockQuote) {
        quoteProducer.send(STOCK_QUOTES_TOPIC, message.symbol.toString(), message)
    }
}

This is it for the producers, pretty simple! This is possible because when using Spring boot and Spring Kafka optimized defaults are applied for any non specified configurations but if you need you can easily override them in configurations or using Spring @Bean definitions.

Let's now build two REST endpoints so we can send messages to those topics using the producers we just created. To pass in the messages we will also create two wrapper DTOs which will be used to pass in data to our REST endpoints.

Create a package called api and a Kotlin class called QuotesController and using constructor depencency injection pass in a reference to our producers, this will be the initial content:

@Controller
@RequestMapping("api")
class QuotesController(val stockQuoteProducer: StockQuoteProducer, val leveragePriceProducer: LeveragePriceProducer) {}

Ok, now let's create a new package under our just create api called dto where we will put our simple DTOs called LeveragePriceDTO and StockQuoteDTO with the following content:

LeveragePriceDTO class:

class LeveragePriceDTO(val symbol: String, val leverage: BigDecimal)

StockQuoteDTO class:

class StockQuoteDTO(val symbol: String, val tradeValue: BigDecimal, @JsonFormat(shape = JsonFormat.Shape.STRING, timezone = "UTC") val isoDateTime: Instant)

This is the package structure after creating these three new classes:

Let's now create two endpoint so we can process and send messages to Kafka, create a new function newQuote inside the QuotesController class:

    @PostMapping("/quotes")
    fun newQuote(@RequestBody stockQuoteDTO: StockQuoteDTO): ResponseEntity<StockQuoteDTO> {
        val stockQuote = StockQuote(stockQuoteDTO.symbol, stockQuoteDTO.tradeValue.toDouble(), stockQuoteDTO.isoDateTime.toEpochMilli())
        stockQuoteProducer.send(stockQuote) // fire and forget
        return ResponseEntity.ok(stockQuoteDTO)
    }

And for Leverage create a new function newLeveragePrice with the following content:

    @PostMapping("/leverage")
    fun newLeveragePrice(@RequestBody leveragePriceDTO: LeveragePriceDTO): ResponseEntity<LeveragePriceDTO> {
        val leveragePrice = LeveragePrice(leveragePriceDTO.symbol, leveragePriceDTO.leverage.toDouble())
        leveragePriceProducer.send(leveragePrice) // fire and forget
        return ResponseEntity.ok(leveragePriceDTO)
    }

Now we need to send some messages to these REST endpoints which will send them to our Kafka topics, in order to do that create a new folder called test-data on the root of the project folder and add a few JSON files with data for leverage and quotes as the examples below:

leveragePrice1.json file:

{
  "symbol": "APPL",
  "leveragePrice": 144.54
}

stockQuote1.json file:

{
  "symbol": "APPL",
  "tradeValue": 123.45,
  "isoDateTime": "2021-11-04T08:22:35.000000Z"
}

Note that we will be receiving an ISO date so make sure to follow the pattern when creating new files.

The structure after creating a few sample data files should be like this:

Now let's create a file called test-data.http on the root of our project where we will have some http calls to invoke our newly created endpoints:

test-data.http file:

POST http://localhost:8080/api/quotes
Content-Type: application/json

< ./test-data/stockQuote1.json

###
POST http://localhost:8080/api/leverage
Content-Type: application/json

< ./test-data/leveragePrice1.json

You can use postman or convert those commands to curl if you prefer. Or if you're using IntelliJ the http files should be automatically recognizable and you can run them directly from the IDE. If you're using VSCode you can install a REST extension which recognizes the http file and you can also run those commands directly from the IDE.

IntelliJ - You can run the calls clicking the green arrow as seen below :

VSCode - You can run the calls clicking the Send Request text on the http file as seen below:

Send some messages to the endpoint and after that let's check the messages on the Broker, I will show how to do that using the Kubernetes setup which is in the project. If you're using docker compose please check the article references on beginning of part I of this series where it's explained in details how to do that using compose.

Checking the messages

First we need to enter the running schema-registry container, so:

List the running pods:

kubectl get pods

you should find the schema-registry pod and copy it's name:

Enter the schema registry container(use the pod name from the command on step 1):

kubectl exec -it schema-registry-54bd4d4b4f-f5wm5 -- bash

Run an avro console consumer to consume the messages from the topics, example below is showing the stock-quotes-topic, replace the topic name to check other topics.

kafka-avro-console-consumer --bootstrap-server kafka:29092 --topic stock-quotes-topic --from-beginning --property "schema.registry.url=http://schema-registry:30081"

After client initialization you should see the messages, I'ves sent some multiple times to illustrate:

A good alternative to using command line and having to enter the running container is to use some Kafka client tooling or kcat. As tooling I really come to enjoy lately using Conduktor which is free for personal use, has a lot of features and is very intuitive to use.

Same messages visualized using Conduktor:

That's it for this post, congratulations if you made it this far, it means you're successfully sending messsages to both topics and we have prepared the ground so we can have some fun using Kafka Streams starting in the next post. Stay tuned.

As usual you can checkout the code up to this stage from the project repo using: git checkout v6

Photo by Adi Goldstein on Unsplash

Spring Kafka Streams playground with Kotlin - I

Marcos Maia — Wed, 09 Feb 2022 20:30:44 +0000

In this series I will cover the creation of a simple SpringBoot Application which demonstrates how to use Spring-kafka to build an application with Kafka Clients including Kafka Admin, Consumer, Producer and Kafka Streams.

The sample application will use the domain of Trading to give it a meaningful context.

I will be posting multiple articles in the next few days with the complete setup.

What are we building

We will build an application that simulates in a very simplified way the processing of quotes and leverage for stocks and sends those quotes in Kafka topics.

quotes -> stock-quotes-topic
leverage-prices -> leverage-prices-topic

We will then create a GlobalKTable with a materialize view for the leverage-prices so we can use it to join with incoming quotes and expose the most up to date leverage for a stock taking advantage of the local default materialized state store Rocksdb that is created for us automatically by Kafka Streams.

I wrote a quick article about Rocksdb in the past if you want more information about it.

With that information we will then create a Kafka Stream that combines such information i.e - when receiving a quote it will try to enrich the quotes with an existing leverage and then will:

Send Apple quotes "APPL" to it's specific topic -> appl-stocks-topic
Send GOOGL quote "GOOGL" to it's specific topic -> googl-stocks-topic
Send all other quotes to a different topic -> all-other-stocks-topic
Count total quotes per type.
Count total quotes per type per interval.

From there we will expose some endpoints to enable us to directly query data from the local storage created by the Kafka streams for leverage and total per interval using grouping, windowing, counting, state stores and Interactive queries from Kafka Streams implementations.

This is a visual overview of what we're building:

In this first part we will create the Spring Boot application, add the files to run local Kafka add the dependencies we need to build the project and the initial schemas.

Prerequisites

You will need Git, Java, Maven, Docker and Docker Compose or Kubernetes installed to follow this series.

I also assume you are familiar with running Spring Boot applications and using docker-compose or Kubernetes to run local infra required. If you are not please follow these other posts where you will find detailed instructions about each of those:

Spring boot crash course - Shows how to create a spring-boot application using the Spring boot initializer site.
One to run them all - Shows how to run Kafka infra with docker-compose and some tricks.
Simplest Spring Kafka Consumer and Producer - Java - Shows how to develop a simple spring-kafka consumer and producer using Java.
Simplest Spring Kafka Consumer and Producer - Kotlin - Shows how to devleop a simple spring-kafka consumer and producer using Kotlin.
Running Kafka on Kubernetes for local development - Shows how to use Kubernetes Kafka infra for local development.

Creating the application

Go to the Spring Boot Starter site and create a project with the following configurations:

Project: Maven Project
Language: Kotlin
Spring Boot: 2.6.3
Dependencies: Spring Web
Project Metadata:

Group: com.maia
Artifact: simple-spring-kafka-stream-kotlin
Name: simple-spring-kafka-stream-kotlin
Description: Spring Kafka Kotlin Playground
Package name: com.maia.simplespringkafkastreamkotlin
Packaging: Jar
Java: 11

If you prefer you can also clone the sample repo of this project using git clone git@github.com:mmaia/simple-spring-kafka-stream-kotlin.git and to check the first step with only the initial app, to do that check out tag v1 git checkout v1 after cloning the project.

Once you create yourself extract the generated project or clone the repo pointing to tag v1 you can then build and run it to validate:

mvn clean package -DskipTests

and then:

mvn spring-boot:run

You should see the app running:

Add local infra

Let's now add the setup to run Kafka locally, I will be providing two setups so you can decide if you want to use docker-compose or Kubernetes. I will not go into details on the implementation as I have past posts covering both approaches as mentioned before:

Docker compose approach is explained in the post [One to run them all]((https://dev.to/thegroo/one-to-run-them-all-1mg6), but in this project we will be using the official Confluent Kafka, schema-registry and zookeeper versions.
Kubernetes approach is explained in the post Running Kafka on Kubernetes for local development

So make sure to add a docker-compose or Kubernetes setup of your preference before continuing to the next steps. You can also simple check out the v2 tag version of the example repository which contain those to take a look if you prefer: git checkout v2.

Run local infra

Again, please check aforementioned articles for further details and explanations if needed.

Docker compose: from a terminal on the project root folder: docker-compose up -d

Kubernetes: open terminal on kubernetes folder and: kind create cluster --config=kind-config.yml and after Kind is started: kubectl apply -f kafka-k8s

Add this line kubernetes/tmp/* to your .gitignore file if you're using kubernetes so git will ignore the mapped volume files for kafka and zookpeeper.

Add Kafka maven dependencies

Make the following changes to the pom.xml file in the root folder of the project.

Add the confluent repository and plugin repository in order to download the confluent dependencies and be able to auto-generate the avro-schemas(more on that later).

<repositories>
    <repository>
    <id>confluent</id>           
        <url>https://packages.confluent.io/maven/</url>
    </repository>
</repositories>

<pluginRepositories>
    <pluginRepository>
    <id>confluent</id>           
        <url>https://packages.confluent.io/maven/</url>
    </pluginRepository>
</pluginRepositories>

Add the version for the libraries we will be using inside the <properties> tag just after the tag:

<avro.version>1.11.0</avro.version>
<confluent.version>7.0.1</confluent.version>
<spring-kafka.version>2.8.2</spring-kafka.version>
<kafka.version>3.0.0</kafka.version>

And finally add the actual library dependencies inside the <dependencies> tag:

<!-- kafka + confluent deps -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>${kafka.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro</artifactId>
    <version>${avro.version}</version>
</dependency>
<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>${confluent.version}</version>
</dependency>
<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-streams-avro-serde</artifactId>
    <version>${confluent.version}</version>
</dependency>

If you prefer you can checkout to the current status using Tag v3: git checkout v3.

Add initial avro schemas

Let's now add the initial Avro Schemas for Leverage and Quotes they will be the "contract" for our messages for the topics where we will initially send Leverage and Quote information as explained in the beginning of this post.

Create a folder under src/main called avro
Create a file inside this folder called: leverage-price-quote.avsc
Create another file inside the same folder called: stock-quote.avsc

Add the following content to those files:

leverage-price-quote.avsc

{
 "namespace": "com.maia.springkafkastreamkotlin.repository",
 "type": "record",
 "name": "LeveragePrice",
 "fields": [
   { "name": "symbol", "type": "string"},
   { "name": "leverage", "type": "double"}
 ]
}

stock-quote.avsc

{
 "namespace": "com.maia.springkafkastreamkotlin.repository",
 "type": "record",
 "name": "StockQuote",
 "fields": [
   { "name": "symbol", "type": "string"},
   { "name": "tradeValue", "type": "double"},
   { "name": "tradeTime", "type": ["null", "long"], "default": null}
 ]
}

In the pom.xml file under the <build><plugins> section add the following avro plugin configuration:

<!-- avro -->
<plugin>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro-maven-plugin</artifactId>
  <version>${avro.version}</version>
  <executions>
    <execution>
      <phase>generate-sources</phase>
      <goals>
    <goal>schema</goal>
      </goals>
    </execution>
  </executions>
</plugin>

The avsc files will be used as a contract definition for the payload you will be sending and consuming from Kafka topics. The confluent Kafka clients enforce that for you. Enforcing a contract can avoid many issues(like poison pills) and together with the Schema Registry provide a framework for consistent incremental versioning of your messages.

Important to mention that the Confluent Schema registry also supports formats other than Avro like Protobuff and Json if you prefer one of them over Avro.

You can now download build the project so the avro plugin will generate the Java Objects we will use in our Producers next.

mvn clean package -DskipTests

This should generate the java objects automatically under: `target/generated-sources/avro/com/maia/springkafkastreamkotlin/repository.

You may git checkout v4 to check out the code up to this point.

Congratulations you have reached the end of the first part of this tutorial series. You can now take a rest and when you're ready move to the second part, which will be published soon.

Where we will be creating the Topics using Kafka Admin Client and also create the Kafka Producers and Rest Endpoints to enable us to create messages and send them to Kafka.

Photo by Joshua Mayo on Unsplash

Running Kafka on Kubernetes for local development with Helm

Marcos Maia — Sun, 30 Jan 2022 09:41:43 +0000

Now that I have posted about running Kafka on Kubernetes using Persistent Volumes(PV) and Persistent Volume Claims(PVC), and another post about running it using Storage Classes it's time to grab that knowledge and have a version of it running with Helm Charts.

Helm is a package manager for Kubernetes, Helm helps you manage, define, install, and upgrade Kubernetes applications. It promotes re-usability and standardization.

In this post we will use Helm to convert the Local Kafka setup using Storage Classes I explained in a previous post. So if you haven't seen that post yet I suggest that you read and play with that setup first as it will help you to understand and extend the approach explained here.

Helm has many more features than the ones covered in this simple post.

As usual you can find the source code with instructions in my Github Repository. To clone it: git clone git@github.com:mmaia/kafka-local-kubernetes.git, have fun.

Pre-reqs, install:

The setup using Helm

In this next section I will just cover the relevant information covering the helm setup and the actual Kubernetes files used are basically the same as the ones covered within the Running Kafka for local development with Storage Class post, so please refer to that article for further details on the Kubernetes setup.

After you clone the repo as described above if you expand the folder helm you will find the following structure:

As seen in my previous posts the kind-config.yaml file is used to setup the local Kubernetes cluster using Kind.

kind.config.yaml file:

apiVersion: kind.x-k8s.io/v1alpha4
kind: Cluster
nodes:
  - role: control-plane
  - role: worker
    extraPortMappings:
      - containerPort: 30092 # internal kafka nodeport
        hostPort: 9092 # port exposed on "host" machine for kafka
      - containerPort: 30081 # internal schema-registry nodeport
        hostPort: 8081 # port exposed on "host" machine for schema-registry
    extraMounts:
      - hostPath: ./tmp
        containerPath: /var/local-path-provisioner
        readOnly: false
        selinuxRelabel: false
        propagation: Bidirectional

Then inside the local-kafka-dev folder you'll find the main Helm Chart file: Chart.yaml and a file to set variables: values.yaml.

Chart.yaml file:

apiVersion: v2 #helm 3
name: local-kafka-dev
appVersion: "7.0.1"
description: A Helm chart for a local, single node Confluent Kafka, Schema Registry and Zookeeper for development.
version: 1.0.0
type: application

values.yaml file:

confluent:
  version: 7.0.1
  kafka: confluentinc/cp-kafka
  schemaRegistry: confluentinc/cp-schema-registry
  zookeeper: confluentinc/cp-zookeeper

Then inside the templates folder you will find all the same Kubernetes files used in the previous post as mentioned before with same minimal changes to use a few of the Helm variables defined in the values.yaml file. The only new file in there is the NOTES.txt which is used to display a nice message after you run the Helm chart as you will be able to see next while running it.

Running it

Open a terminal and cd to helm folder
Start Kind running: kind create cluster --config=kind-config.yml
Run helm install ${pick_a_name} local-kafka-dev. i.e - helm install local-kafka local-kafka-dev
When done stop kafka setup using helm uninstall ${name_picked_step_3}, you may also stop Kind if you want: kind delete cluster

That's it, it's done, you have a functional local Kafka +
Schema Registry running on Kubernetes and managed with Helm Charts which you can reach from your application running on your developer machine or IDE.

Enjoy!

Photo by ThisisEngineering RAEng on Unsplash

Running Kafka on Kubernetes for local development with Storage class

Marcos Maia — Mon, 17 Jan 2022 14:13:08 +0000

In a recent post I showed a setup to run Kafka on Kubernetes locally using Persistent Volumes(PV) and Persistent Volume Claims(PVC), this post covers the setup using Kubernetes Storage Class(SC) and leveraging the default one provisioned automatically by Kind which is the Rancher local-path-provisioner.

This setup is simpler and uses less code than the previous one with the trade off of having a bit less control over the path of data externalized to the host machine while requiring some internal knowledge of Kind to set it up. In general I favor this approach for local development over the one from my previous post.

Strimzi is an awesome simpler alternative to achieve the same, check it out. My goal with this setup here is for learning and to have a more "realistic" Kubernetes setup on local development machine so I opted to not use Strimzi or Helm charts.

I have created and tested these approaches on a Linux Development machine. It should work for Mac and Windows with some minimal adjustments also but I have never tried it.

You can get the full source from Github repo where you will find the files and Quick Start for both aforementioned approaches. To clone the repo git clone git@github.com:mmaia/kafka-local-kubernetes.git.

Pre-reqs, install:

The setup using Storage class

If you checked out the repo described above the setup presented here is under storage-class-setup folder. You will find multiple Kubernetes declarative files in this folder, please notice that you could also combine all files in a single one separating them with a line containing triple dashes ---, if combining them is your preference you can open a terminal and from the storage-class-setup folder run for each in ./kafka-k8s/*; do cat $each; echo "---"; done > local-kafka-combined.yaml this will concatenate all files in a single one called local-kafka-combined.yaml.

The main thing to notice in this setup below compared to the previous one is that you don't have any PV or PVC configuration files this time because we're leveraging the default Rancher local-path-provisioner provided by Kind automatically through it's default Storage class.

I keep them separate to explicitly separate each type in this case and because it's convenient as you can just run kubectl pointing to the directory as described below in the "Running it" section.

kind-config.yaml - This file configures Kind to expose the kafka and schema-registry ports to the local machine host so you can connect your application while developing from your IDE or command line and connect with Kafka running on Kubernetes it also maps the default path of the Rancher storage provisioner from Kind container to your local host machine.

apiVersion: kind.x-k8s.io/v1alpha4
kind: Cluster
nodes:
  - role: control-plane
  - role: worker
    extraPortMappings:
      - containerPort: 30092 # internal kafka nodeport
        hostPort: 9092 # port exposed on "host" machine for kafka
      - containerPort: 30081 # internal schema-registry nodeport
        hostPort: 8081 # port exposed on "host" machine for schema-registry
    extraMounts:
      - hostPath: ./tmp
        containerPath: /var/local-path-provisioner
        readOnly: false
        selinuxRelabel: false
        propagation: Bidirectional

kafka-network-np.yaml - Sets up the internal Kubernetes network used by the setup.

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: kafka-network
spec:
  ingress:
    - from:
        - podSelector:
            matchLabels:
              network/kafka-network: "true"
  podSelector:
    matchLabels:
      network/kafka-network: "true"

kafka-service.yaml - This file defines the mappings between the internal containers and ports that are exposed, called NodePorts by defaul in Kubernetes nodeports can be used in the range 30000 to 32767.

apiVersion: v1
kind: Service
metadata:
  labels:
    service: kafka
  name: kafka
spec:
  selector:
    service: kafka
  ports:
    - name: internal
      port: 29092
      targetPort: 29092
    - name: external
      port: 30092
      targetPort: 9092
      nodePort: 30092
  type: NodePort

kafka-ss.yaml - This is the definition of Kafka in this setup, this time we use a Stateful Set.

apiVersion: apps/v1
kind: StatefulSet
metadata:
  labels:
    service: kafka
  name: kafka
spec:
  serviceName: kafka
  replicas: 1
  selector:
    matchLabels:
      service: kafka
  template:
    metadata:
      labels:
        network/kafka-network: "true"
        service: kafka
    spec:
      enableServiceLinks: false
      containers:
      - name: kafka
        imagePullPolicy: IfNotPresent
        image: confluentinc/cp-kafka:7.0.1
        ports:
          - containerPort: 29092
          - containerPort: 9092
        env:
          - name: CONFLUENT_SUPPORT_CUSTOMER_ID
            value: "anonymous"
          - name: KAFKA_ADVERTISED_LISTENERS
            value: "INTERNAL://kafka:29092,LISTENER_EXTERNAL://kafka:9092"
          - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE
            value: "true"
          - name: KAFKA_BROKER_ID
            value: "1"
          - name: KAFKA_DEFAULT_REPLICATION_FACTOR
            value: "1"
          - name: KAFKA_INTER_BROKER_LISTENER_NAME
            value: "INTERNAL"
          - name: KAFKA_LISTENERS
            value: "INTERNAL://:29092,LISTENER_EXTERNAL://:9092"
          - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
            value: "INTERNAL:PLAINTEXT,LISTENER_EXTERNAL:PLAINTEXT"
          - name: KAFKA_NUM_PARTITIONS
            value: "1"
          - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR
            value: "1"
          - name: KAFKA_LOG_CLEANUP_POLICY
            value: "compact"
          - name: KAFKA_ZOOKEEPER_CONNECT
            value: "zookeeper:2181"
        resources: {}
        volumeMounts:
          - mountPath: /var/lib/kafka/data
            name: kafka-data
      hostname: kafka
      restartPolicy: Always
  volumeClaimTemplates:
    - metadata:
        name: kafka-data
      spec:
        accessModes:
          - ReadWriteOnce
        resources:
          requests:
            storage: 1Gi

The remaining files are declarative Kubernetes configurations files to schema-registry and zookeeper.

schema-registry-deployment.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    service: schema-registry
  name: schema-registry
spec:
  replicas: 1
  selector:
    matchLabels:
      service: schema-registry
  strategy: {}
  template:
    metadata:
      labels:
        network/kafka-network: "true"
        service: schema-registry
    spec:
      enableServiceLinks: false
      containers:
        - env:
            - name: SCHEMA_REGISTRY_HOST_NAME
              value: "schema-registry"
            - name: SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS
              value: "kafka:29092"
            - name: SCHEMA_REGISTRY_LISTENERS
              value: "http://0.0.0.0:30081"
          image: confluentinc/cp-schema-registry:7.0.1
          name: schema-registry
          ports:
            - containerPort: 30081
          resources: {}
      hostname: schema-registry
      restartPolicy: Always

schema-registry-service.yaml

apiVersion: v1
kind: Service
metadata:
  labels:
    service: schema-registry
  name: schema-registry
spec:
  ports:
    - port: 30081
      name: outport
      targetPort: 30081
      nodePort: 30081
  type: NodePort
  selector:
    service: schema-registry

zookeeper-service.yaml

apiVersion: v1
kind: Service
metadata:
  labels:
    service: zookeeper
  name: zookeeper
spec:
  ports:
    - name: "2181"
      port: 2181
      targetPort: 2181
  selector:
    service: zookeeper

zookeeper-ss.yaml - Again the main difference this time is the usage of Stateful Set.

apiVersion: apps/v1
kind: StatefulSet
metadata:
  labels:
    service: zookeeper
  name: zookeeper
spec:
  serviceName: zookeeper
  replicas: 1
  selector:
    matchLabels:
      service: zookeeper
  template:
    metadata:
      labels:
        network/kafka-network: "true"
        service: zookeeper
    spec:
      enableServiceLinks: false
      containers:
        - name: zookeeper
          imagePullPolicy: IfNotPresent
          image: confluentinc/cp-zookeeper:7.0.1
          ports:
            - containerPort: 2181
          env:
            - name: ZOOKEEPER_CLIENT_PORT
              value: "2181"
            - name: ZOOKEEPER_DATA_DIR
              value: "/var/lib/zookeeper/data"
            - name: ZOOKEEPER_LOG_DIR
              value: "/var/lib/zookeeper/log"
            - name: ZOOKEEPER_SERVER_ID
              value: "1"
          resources: {}
          volumeMounts:
            - mountPath: /var/lib/zookeeper/data
              name: zookeeper-data
            - mountPath: /var/lib/zookeeper/log
              name: zookeeper-log
      hostname: zookeeper
      restartPolicy: Always
  volumeClaimTemplates:
    - metadata:
        name: zookeeper-data
      spec:
        accessModes:
          - ReadWriteOnce
        resources:
          requests:
            storage: 250Mi
    - metadata:
        name: zookeeper-log
      spec:
        accessModes:
          - ReadWriteOnce
        resources:
          requests:
            storage: 250Mi

Running it

Open a terminal and cd to the storage-class-setup folder.
Create a folder called "tmp", this is where the storage will be automatically provisioned by the default Kind storage class.
Run kind specifying configuration: kind create cluster --config=kind-config.yml. This will start a Kind kubernetes control plane + worker.
Run kubernetes configuration for kafka kubectl apply -f kafka-k8s
When done stop kubernetes objects: kubectl delete -f kafka-k8s and then if you want also stop the kind cluster which will also delete the storage on the host machine: kind delete cluster.

After running the kubectl apply command(step 4 above) check your local tmp folder where you will find the automated storage mapped to your local host disk, notice that those folders will be deleted when you shutdown the Kind cluster but they will persist over pod restarts of Kafka and zookeeper.

Connecting a Kafka client

I decided to add this section because I commonly see developers struggling to connect their Kafka clients from the IDE or local dev machine to a local Kafka running on docker, docker-compose or kubernetes, mostly because the client gives an error that it can't find host kafka when running.

There are many ways to solve that I will explain a few ways here.

The simplest way to do it is to add kafka and schema-registry to your /etc/hosts file in the host machine, so when the broker configures LISTENER_EXTERNAL://kafka:9092 which will be advertised to the kafka client the client will resolve kafka to localhost and it will just work. In some configurations you do the same with schema-registry. Some people don't like to do this so there are options.
Change the configuration on your compose file or kubernetes kafka config file to LISTENER_EXTERNAL://localhost:9092, so now when in your local client you specify kafka broker address to localhost:9092 it will just work as it will match the advertised address the client received from the broker. The downside of doing this is that if you decide to package your application to run locally in a container inside the docker-compose or kubernetes network the call will fail as localhost will then point to the application container / pod instead of kafka.
When using Kubernetes use a tool like Telepresence to expose and route kubernetes ports.
When using docker-compose you might leverage the docker internal address mappings as describe nicely in this article.

You can find a detailed explanation on why this happens in this Confluent blog post.

That's it, it's done, you have a functional local Kafka +
Schema Registry running on Kubernetes that you can reach from your application running on your developer machine or IDE.

Photo by Fotis Fotopoulos on Unsplash

Running Kafka on kubernetes for local development

Marcos Maia — Tue, 11 Jan 2022 20:02:57 +0000

In this post I will cover the steps to run Kafka locally on your development machine using Kubernetes. Usually I go with docker-compose for that as it is simpler to get it going but due to the omnipresence of Kubernetes these days in companies I decided to also port my local Confluent Kafka setup to run on Kubernetes for practicing and getting closer to production environments where my application will most likely also run on Kubernetes.

If you don't need strict control over the Kafka and Zookeeper storage locations on your developer host machine which is most of the time the case, I would recommend that you use the Storage class setup approach because it's simpler and uses less code as it leverages the default storage provider used by Kind.

This is a minimal local Kafka setup meant for development so it's a single instance of Kafka, Schema-registry and Zookeeper. You may want to run a local kafka for quick and easy test and prototyping and also to save money from running it in the cloud just for functional development, if you combine that with also packaging and running your application on Kubernetes you have a very efficient setup that closely mirrors cloud environments where your app and the Kubernetes cluster you will use for the environments are probably running.

Strimzi is an awesome simpler alternative to achieve the same, check it out. My goal with this setup here is for learning and to have a more "realistic" Kubernetes setup on local development machine so I opted to not use Strimzi or Helm charts.

In the past couple of days I came up with two local setups running Kafka, Schema-registry and Zookeeper on local development machine with Kubernetes using Kind, in this first post I will cover a setup using Persistent Volumes and Persistent Volume Claims and in the next one I will cover using Storage classes.

I have created and tested these approaches on a Linux Development machine. It should work for Mac and Windows also but I have never tried it.

Well I guess this is more than enough introduction let's have some fun.

Pre-reqs, install:

The setup using Persistent Volumes and Persistent Volume Claims

If you checked out the repo described above the setup presented here is under pv-pvc-setup folder. You will find multiple Kubernetes declarative files in this folder, please notice that you could also combine all files in a single one separating them with a line containing triple dashes ---, if combining them is your preference you can open a terminal and from the pv-pvc-setup folder run for each in ./kafka-k8s/*; do cat $each; echo "---"; done > local-kafka-combined.yaml this will concatenate all files in a single one called local-kafka-combined.yaml.

I keep them separate to explicitly separate each type in this case and because it's convenient as you can just run kubectl pointing to the directory as described below in the "Running it" section.

apiVersion: kind.x-k8s.io/v1alpha4
kind: Cluster
nodes:
  - role: control-plane
  - role: worker
    extraPortMappings:
      - containerPort: 30092 # internal kafka nodeport
        hostPort: 9092 # port exposed on "host" machine for kafka
      - containerPort: 30081 # internal schema-registry nodeport
        hostPort: 8081 # port exposed on "host" machine for schema-registry
    extraMounts:
      - hostPath: ./tmp/kafka-data
        containerPath: /var/lib/kafka/data
        readOnly: false
        selinuxRelabel: false
        propagation: Bidirectional
      - hostPath: ./tmp/zookeeper-data/data
        containerPath: /var/lib/zookeeper/data
        readOnly: false
        selinuxRelabel: false
        propagation: Bidirectional
      - hostPath: ./tmp/zookeeper-data/log
        containerPath: /var/lib/zookeeper/log
        readOnly: false
        selinuxRelabel: false
        propagation: Bidirectional

Notice the mapping from the internal container paths to the external hostPath on the local machine. The local paths will need to be manually created before running the setup as per instructions on the section "Running it" below.

This is it for the Kind configuration now let's check the Kubernetes files(under kafka-kl8s if you checked out the project):

kafka-deployment.yaml - Configures the kafka broker and exposes an internal(for kubernetes network) and external port(for kafka clients) for kafka also maps an internal volume to expose kafka data files.

apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    service: kafka
  name: kafka
spec:
  replicas: 1
  selector:
    matchLabels:
      service: kafka
  strategy:
    type: Recreate
  template:
    metadata:
      labels:
        network/kafka-network: "true"
        service: kafka
    spec:
      enableServiceLinks: false
      containers:
      - name: kafka
        imagePullPolicy: IfNotPresent
        image: confluentinc/cp-kafka:7.0.1
        ports:
          - containerPort: 29092
          - containerPort: 9092
        env:
          - name: CONFLUENT_SUPPORT_CUSTOMER_ID
            value: "anonymous"
          - name: KAFKA_ADVERTISED_LISTENERS
            value: "INTERNAL://kafka:29092,LISTENER_EXTERNAL://kafka:9092"
          - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE
            value: "true"
          - name: KAFKA_BROKER_ID
            value: "1"
          - name: KAFKA_DEFAULT_REPLICATION_FACTOR
            value: "1"
          - name: KAFKA_INTER_BROKER_LISTENER_NAME
            value: "INTERNAL"
          - name: KAFKA_LISTENERS
            value: "INTERNAL://:29092,LISTENER_EXTERNAL://:9092"
          - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
            value: "INTERNAL:PLAINTEXT,LISTENER_EXTERNAL:PLAINTEXT"
          - name: KAFKA_NUM_PARTITIONS
            value: "1"
          - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR
            value: "1"
          - name: KAFKA_LOG_CLEANUP_POLICY
            value: "compact"
          - name: KAFKA_ZOOKEEPER_CONNECT
            value: "zookeeper:2181"
        resources: {}
        volumeMounts:
          - mountPath: /var/lib/kafka/data
            name: kafka-data
      hostname: kafka
      restartPolicy: Always
      volumes:
        - name: kafka-data
          persistentVolumeClaim:
            claimName: kafka-pvc

kafka-network-np.yaml - Sets up the internal Kubernetes network used by the setup.

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: kafka-network
spec:
  ingress:
    - from:
        - podSelector:
            matchLabels:
              network/kafka-network: "true"
  podSelector:
    matchLabels:
      network/kafka-network: "true"

kafka-pv.yaml - this file describes the persistent volume used.

apiVersion: v1
kind: PersistentVolume
metadata:
  name: kafka-pv
spec:
  accessModes:
    - ReadWriteOnce
  storageClassName: kafka-local-storage
  capacity:
    storage: 5Gi
  persistentVolumeReclaimPolicy: Retain
  hostPath:
    path: /var/lib/kafka/data

kafka-pvc.yaml - this file is the claim used by the 'pod' as described in the deployment file for kafka above and associated to the persistent volume.

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: kafka-pvc
spec:
  accessModes:
    - ReadWriteOnce
  storageClassName: kafka-local-storage
  resources:
    requests:
      storage: 5Gi

kafka-service.yaml - This file defines the mappings between the internal containers and ports that are exposed, called NodePorts in Kubernetes.

apiVersion: v1
kind: Service
metadata:
  labels:
    service: kafka
  name: kafka
spec:
  selector:
    service: kafka
  ports:
    - name: internal
      port: 29092
      targetPort: 29092
    - name: external
      port: 30092
      targetPort: 9092
      nodePort: 30092
  type: NodePort

The remaining files are declarative Kubernetes configurations files to schema-registry and zookeeper.

schema-registry-deployment.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    service: schema-registry
  name: schema-registry
spec:
  replicas: 1
  selector:
    matchLabels:
      service: schema-registry
  strategy: {}
  template:
    metadata:
      labels:
        network/kafka-network: "true"
        service: schema-registry
    spec:
      enableServiceLinks: false
      containers:
        - env:
            - name: SCHEMA_REGISTRY_HOST_NAME
              value: "schema-registry"
            - name: SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS
              value: "kafka:29092"
            - name: SCHEMA_REGISTRY_LISTENERS
              value: "http://0.0.0.0:30081"
          image: confluentinc/cp-schema-registry:7.0.1
          name: schema-registry
          ports:
            - containerPort: 30081
          resources: {}
      hostname: schema-registry
      restartPolicy: Always

schema-registry-service.yaml

apiVersion: v1
kind: Service
metadata:
  labels:
    service: schema-registry
  name: schema-registry
spec:
  ports:
    - port: 30081
      name: outport
      targetPort: 30081
      nodePort: 30081
  type: NodePort
  selector:
    service: schema-registry

zookeeper-data-pv.yaml

apiVersion: v1
kind: PersistentVolume
metadata:
  name: zookeeper-data-pv
spec:
  accessModes:
    - ReadWriteOnce
  storageClassName: zookeeper-data-local-storage
  capacity:
    storage: 5Gi
  persistentVolumeReclaimPolicy: Retain
  hostPath:
    path: /var/lib/zookeeper/data

zookeeper-data-pvc.yaml

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: zookeeper-data-pvc
spec:
  accessModes:
    - ReadWriteOnce
  storageClassName: zookeeper-data-local-storage
  resources:
    requests:
      storage: 5Gi

zookeeper-deployment.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    service: zookeeper
  name: zookeeper
spec:
  replicas: 1
  selector:
    matchLabels:
      service: zookeeper
  strategy: {}
  template:
    metadata:
      labels:
        network/kafka-network: "true"
        service: zookeeper
    spec:
      containers:
        - env:
            - name: TZ
            - name: ZOOKEEPER_CLIENT_PORT
              value: "2181"
            - name: ZOOKEEPER_DATA_DIR
              value: "/var/lib/zookeeper/data"
            - name: ZOOKEEPER_LOG_DIR
              value: "/var/lib/zookeeper/log"
            - name: ZOOKEEPER_SERVER_ID
              value: "1"
          image: confluentinc/cp-zookeeper:7.0.1
          name: zookeeper
          ports:
            - containerPort: 2181
          resources: {}
          volumeMounts:
            - mountPath: /var/lib/zookeeper/data
              name: zookeeper-data
            - mountPath: /var/lib/zookeeper/log
              name: zookeeper-log
      hostname: zookeeper
      restartPolicy: Always
      volumes:
        - name: zookeeper-data
          persistentVolumeClaim:
            claimName: zookeeper-data-pvc
        - name: zookeeper-log
          persistentVolumeClaim:
            claimName: zookeeper-log-pvc

zookeeper-log-pv.yaml

apiVersion: v1
kind: PersistentVolume
metadata:
  name: zookeeper-log-pv
spec:
  accessModes:
    - ReadWriteOnce
  storageClassName: zookeeper-log-local-storage
  capacity:
    storage: 5Gi
  persistentVolumeReclaimPolicy: Retain
  hostPath:
    path: /var/lib/zookeeper/log

zoopeeker-log-pvc.yaml

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: zookeeper-log-pvc
spec:
  accessModes:
    - ReadWriteOnce
  storageClassName: zookeeper-log-local-storage
  resources:
    requests:
      storage: 5Gi

zookeeper-service.yaml

apiVersion: v1
kind: Service
metadata:
  labels:
    service: zookeeper
  name: zookeeper
spec:
  ports:
    - name: "2181"
      port: 2181
      targetPort: 2181
  selector:
    service: zookeeper

Running it

After cloning the project open a terminal and cd to the pv-pvc-setup folder. Or if you're creating the files navigate to folder where the kind-config.yaml is located.
Create the folders on your local host machine so the persistent volumes can be persisted to the file system and you
will be able to restart the kafka and zookeeper cluster without loosing data from topic. Restarting the kind cluster
will delete the contents of persistent volumes, this is by design how Kind works when having
a propagation: Bidirectional configuration. To create the folders you'll need to have uid and gid from same
user running kind cluster otherwise the persistent folders will not be properly persisted. i.e - make sure to
create tmp/kafka-data, tmp/zookeeper-data/data and tmp/zookeeper-data/log
from same level where the kind-config.yaml file you're running kind with or it won't work as expected.
Run kind specifying configuration: kind create cluster --config=kind-config.yml. This will start a kubernetes
control plane + worker. Check Kind docker containers with Kubernetes control-plane and worker running docker ps.
Run kubernetes configuration for kafka kubectl apply -f kafka-k8s
When done stop kubernetes objects: kubectl delete -f kafka-k8s and then if you want also stop the kind cluster, be aware that this will also delete the local storage on the host machine: kind delete cluster

If you have problems to connect your kafka client from your local development machine to the broker running in docker or Kubernetes please check the section Connecting a Kafka client on the end of this other post where you'll find details on how to fix it and links explaining why it happens and sort this out completely, have fun.

That's all for now folks.

Stay tunned as I will post about a simpler approach using the default Storage Class automatically provisioned by Kind (Rancher/local-path-provisioner) which simplifies the setup considerably with the trade off of not having so much control over the host local storage where kafka and zookpeeper files are stored on the host machine.

Photo by Fotis Fotopoulos on Unsplash

No BS guide to Kubernetes

Marcos Maia — Thu, 30 Dec 2021 23:49:56 +0000

Recently due to the huge presence and dominance of Kubernetes as an orchestration tool I decided to take another look at it. I have worked a little bit with Kubernetes back in 2017 but having mostly focused on Development after that I didn't do much with it since then.

Going back to it was a bit of a pain initially as there's so much material around and so much information that it took me longer than I wanted to have a simple initial example setup project with Kubernetes running in my local environment. To mitigate that for others I decided to do this quick article about it.

The focus of this article is running a minimal Kubernetes local setup so you can get a first hands on experience with it. I do not explain the concepts of Kubernetes or Docker in this post.

I plan in the near future to migrate some of my local docker-compose setups to Kubernetes for fun and learning pourposes, so stay tuned, enough intro for now, let's go for it.

Pre-requisites:

Install:

Create kind configuration file and start kind

Create a cofiguration file for kind: kind-config.yml

apiVersion: kind.x-k8s.io/v1alpha4
kind: Cluster
nodes:
- role: control-plane
  extraPortMappings:
  - containerPort: 31111
    hostPort: 31111
    listenAddress: "0.0.0.0" # Optional, defaults to "0.0.0.0"
    protocol: tcp # Optional, defaults to tcp
- role: worker

Now start kind using the config file: kind create cluster --config=kind-config.yml this will start a local docker container containing the Kind kubernetes control plane we will be using, it also exposes a NodePort to your host so you can reach the nginx service we will run in kubernetes next.

Note: This file(kind-config.yml) is required to run local kubernetes Nodeport as described here

Kubernetes configuration files

Create the kubernetes declarative files:

A simple app running in a docker container wrapped in a pod.

Let's use nginx hello demo image. Create a file called:

hello-pod.yml

apiVersion: v1
kind: Pod
metadata:
  name: hello-pod
  labels:
    app: nginx
spec:
  containers:
    - name: web-nginx
      image: nginxdemos/hello:latest
      ports:
        - containerPort: 80

A kubernetes nodeport service, create a file called:

svc-config.yml

apiVersion: v1
kind: Service
metadata:
  name: nginx-nodeport
spec:
  type: NodePort
  ports:
  - port: 8000
    targetPort: 80
    nodePort: 31111
    protocol: TCP
  selector:
    app: nginx

Apply configurations:

kubectl apply -f hello-pod.yml

kubectl apply -f svc-config.yml

You can optionally apply multiple files pointing to a directory so instead of the two commands above you could place those 2 files in a folder for example k8s and then run:

kubectl apply -f k8s

Access your running kubernetes setup application, on browser navigate to :

http://localhost You should see it:

Some useful commands:

docker ps -> show running docker containers used by kind to create the kubernetes control plane and worker.

kubectl cluster-info -> display kind cluster info.

kubectl describe svc $svc_name -> describes service

kubectl get pods -> show pod information

kubectl get pods --show-labels

kubectl version --short

kind get clusters -> display kluster names

Delete commands:

kubectl delete pod $pod_name

kubectl delete svc $service_name

kind delete cluster $cluster_name

*This is it, no BS.... it's done! *

AS usual the code and sketch for this article can be found on Github repo

Photo by Ian Taylor on Unsplash

JHipster Crash Course

Marcos Maia — Sat, 08 Feb 2020 13:42:07 +0000

In this post, I will cover and guide you through the necessary steps to generate a JHipster application. JHipster is an opinionated generator that helps you to quickly bootstrap projects using Springboot and different front end frameworks and technology stack. I use it for some of my side projects and to quickly bootstrap applications to study new technologies and prototype solutions. Jhipster is quite flexible and gives you many options, in this and next posts I will write about my most common preferences and customizations when using Jhipster.

For this series, I am picking a JHipster generated microservices-gateway project to start, if you decide to go with a monolithic approach in the following steps most of the customizations and instructions will also apply.

So if you are starting a new serious project and you are not studying distributed microservices systems or isn't familiar with them it is better that you pick a monolithic approach instead when selecting the available Jhipster options, picking that option will not change much from the customizations that I will explain in the next posts here as the monolithic version of the JHipster project has many similarities with the Gateway project I generate in this post.

Jhipster has good documentation so if you find yourself lost here, take a look at their official documentation as your doubts will most likely be already answered there. The version of Jhipster I am using is 6.7.0 for this post.

Generate a Jhipster project

The first step to generate a Jhipster project is to install all its required dependencies and CLI as explained in their Getting Started guide, to follow up in this post please make sure you have your machine properly set up following the instructions on the link.

I also have some posts which you can follow if you want to have a basic setup including docker, java or getting your initial overview of Springboot applications.

Jhipster is a generic generator which enables you to bootstrap your projects and pick many different technology stacks and has many options, what I am doing in this post is guiding you with my opinionated tech stack and choices I like the best for my side projects when using Jhipster and the customizations I normally do to fit my preferences.

Open a terminal window, navigate to the personal working folder of your preference and create an initial folder to host the project:

mkdir jhipster-custom && cd jhipster-custom

Start the jhipster generator in the command line:

jhipster

You will be presented with a list of options to pick, choose Microservice gateway or Monolithic application

Next step is to choose a base name for your application: CustomJhipster
Now you will be prompted to select a port, leave the default: 8080
Define a java package name: io.stockgeeks
Pick a service discovery option(if you pick the monolithic app option, most likely this one will not be there), it is safe to skip it in this case, if you picked Gateway Application on item 2 select: Jhipster Registry which uses Eureka with Spring Cloud Config support.
Type of authentication: JWT authentication (stateless, with a token).
Database, pick: SQL and then select Postgres and Postgres again in the development database option.
Cache options pick: Yes, with the Ehcache implementation.
Use hibernate 2nd level cache: Yes
Maven or Gradle: Maven
Other technologies: None, just hit enter without selecting any for now.
Front End framework: Angular
Bootswatch theme: Default JHipster
Internationalization support: No lets stick to Englis for the series.
Testing Frameworks: None just hit enter and skip it, JUnit is already there in the generated project.
Other generators from Marketplace: No

After this last selection the project will be generated for you, if it fails, please review and make sure to follow the links above to set up your environment properly.

It will take some time but finally, you should see a message like: INFO! Congratulations, JHipster execution is complete!. Awesome you have generated your JHipster project, lets now run the initial generated application before we start customizing it.

All commands to run the application next should be executed from the level of the project folder we created initially.

To run the generated project you will first need to run the Postgresql database so your application can connect to it, luckily Jhipster is shipped with a docker-compose configuration for that which is under, the initial default location is from the root of the project: src/main/docker/postgresql.yml so let's start it, from the initial folder on the generated project: docker-compose -f src/main/docker/postgresql.yml up -d, after running the command wait a few seconds and check if PostgreSQL is indeed started: docker ps you should see a running PostgreSQL database running:

If you're running a monolithic version of Jhipster app, this should be enough to run the actual Springboot application, in case you're running the Gateway version as I did you will also need to have the service registry running before starting the application, there is also a docker-compose available for that, run: docker-compose -f src/main/docker/jhipster-registry.yml up -d, wait a few seconds and check again docker ps, you should now see the registry also running + Postgres.

Now that Postgres and the registry are running, let's run the application. from the initial project folder:

mvn spring-boot:run

Being the first time we run the project, it will build, download dependencies and should start after some time, you should then see something like this:

Started CustomJhipsterApp in 8.199 seconds (JVM running for 8.607)

Application 'CustomJhipster' is running! Access URLs:
Local:      http://localhost:8080/
External:   http://127.0.1.1:8080/
Profile(s):     [dev, swagger]

Navigate in your browser to http://localhost:8080 and login in the generated app with admin/admin initial default credentials.

Navigate around to see the default generated application and what it offers, its a lot that you get out of the box.

The generated application from this post is available at Github, you can clone it with: git clone git@github.com:stockgeeks/jhipster-custom.git && git checkout tags/v1.0, after the project is cloned and the tagged version checked out by the previous command you should have the exactly version I generated while creating this post.

There are many ways to evolve your generated Jhipster application, they provide nice tooling and CLI for it and you can also use the Angular CLI for front end among many other possibilities. I use it as a kickstart for my side projects but I have many personal preferences and from the initial generated application I usually do a lot of small customization to fit my preferred development setup.

Stay tuned. Feedback is highly appreciated.

References

JHipster Website

Photo by David Rangel on Unsplash

That's it for now, hope you liked it. Keep learning!

Follow me on Twitter

Simple RocksDB with Java - Crash Course

Marcos Maia — Sat, 28 Dec 2019 14:16:32 +0000

In this post, we are going to implement a very simple client persistence key-value store with RocksDB. The idea is to have the simplest possible implementation as an initial reference.

You can find the link with full source code for this post in the references section in the end of this post.

From the RocksDB documentation: RocksDB is a persistent and embeddable key-value store for fast storage environments. Its open-source and was created and is still maintained by Facebook and it's optimized for fast, low latency storage such as flash drives and high-speed disk drives.

Let's create our example, first step is to generate an initial SpringBoot Application with basic REST endpoints, navigate to start.spring.io and create an application, select Java 11 adding actuator and web as dependencies to start with, if you have never done so, check out this post where you'll find out how to do it in very simple steps. Let's call this app rocksdbBootApp.

Once the app is created the next step is to add Rocksjava as a dependency, here you see a snippet of how to do it using maven.



<properties>
  <java.version>11</java.version>
  <rocksdb.version>5.5.1</rocksdb.version>
</properties>
<dependency>
  <groupId>org.rocksdb</groupId>
  <artifactId>rocksdbjni</artifactId>
  <version>${rocksdb.version}</version>
</dependency>

Let's now create an interface with the signature of the basic operations we want, KeyValueRepository.



package io.stockgeeks.repository;

public interface KeyValueRepository<K, V> {
  void save(K key, V value);
  V find(K key);
  void delete(K key);
}

Nice, we have basic initial operations to save, find and delete entries, let's implement these basic operations, the most important thing to notice in the following fragments of code is that RocksDB works with bytes so it's all transformed to byte array and back when interacting with it's APIs, because we're using Strings here in this simple example we can simply use the getBytes to transform a String into a byte array and build it back using String(byte[] bytes) constructor. If you need to serialize objects you can use spring SerializationUtils or if you're not using Spring there's also a SerializationUtils class in apache-commons API that can be used.

First we implement the interface and we declare 1 constant with the local storage name which will be reflected in a directory in the file system where RocksDB will keep the data structures, we also define the java File reference which points to the final folder structure and the RocksDB reference we're going to use, we mark the class as a @Repository with the Spring annotation and add logging using lombok @Slf4j annotation:



@Slf4j
@Repository
public class RocksDBRepositoryImpl implements KeyValueRepository<String, String> {

  private final static String NAME = "first-db";
  File dbDir;
  RocksDB db;

Let's now create an initialization method that will initialize the RocksDB file system structure and it's configurations and get it ready for interactions.

We use a @PostConstruct annotation so this fragment will be executed after the application starts. We then make sure the file system directory structure that we need is created and we open the RocksDB after the initialization, if you follow the code here as it is the file system structure where it will be maintained is under: /tmp/rocks-db/first-db, you will probably need to adjust the code paths a bit if you're using a Windows environment.

Note: Java has a universal path



  @PostConstruct
  void initialize() {
    RocksDB.loadLibrary();
    final Options options = new Options();
    options.setCreateIfMissing(true);
    dbDir = new File("/tmp/rocks-db", NAME);
    try {
      Files.createDirectories(dbDir.getParentFile().toPath());
      Files.createDirectories(dbDir.getAbsoluteFile().toPath());
      db = RocksDB.open(options, dbDir.getAbsolutePath());
    } catch(IOException | RocksDBException ex) {
      log.error("Error initializng RocksDB, check configurations and permissions, exception: {}, message: {}, stackTrace: {}",
        ex.getCause(), ex.getMessage(), ex.getStackTrace());
    }
    log.info("RocksDB initialized and ready to use");
  }

With the application ready to go we can now have the basic operations methods implemented, starting with the operation to save, nothing really special about it, just a simple straightforward implementation, we log the method call and we use the RocksDB.save method converting the Strings passed as aforementioned, in case of any exceptions we simply log the error in this case:



    @Override
  public synchronized void save(String key, String value) {
    log.info("save");
    try {
      db.put(key.getBytes(), value.getBytes());
    } catch (RocksDBException e) {
      log.error("Error saving entry in RocksDB, cause: {}, message: {}", e.getCause(), e.getMessage());
    }
  }

The find operation is also straightforward, the only important thing to notice is that we check the result for null avoiding a NullpointerException during runtime in case the passed in key does not exist and the aforementioned conversion from byte array to String using the String constructor.



  @Override
  public String find(String key) {
    log.info("find");
    String result = null;
    try {
      byte[] bytes = db.get(key.getBytes());
      if(bytes == null) return null;
      result = new String(bytes);
    } catch (RocksDBException e) {
            log.error("Error retrieving the entry in RocksDB from key: {}, cause: {}, message: {}", key, e.getCause(), e.getMessage());
    }
    return result;
  }

The deletion follows the same pattern and its implementation is straightforward.



  @Override
  public void delete(String key) {
    log.info("delete");
    try {
      db.delete(key.getBytes());
    } catch (RocksDBException e) {
            log.error("Error deleting entry in RocksDB, cause: {}, message: {}", e.getCause(), e.getMessage());
    }
  }
}

Now that we have the basic implementation in place we're good to create an API to complete the example in a way that is more interactive, so let's create simple API using Spring REST, the API simply pass in the received values to the RocksDB implementation exposing it using basic HTTP method calls and handle the results returning 200 or 204 when applicable, please check Spring Boot Crash Course for more details, explanation, and links about building a basic API with SpringBoot if you feel like you need further references:



public class RocksApi {

  private final KeyValueRepository<String, String> rocksDB;

  public RocksApi(KeyValueRepository<String, String> rocksDB) {
    this.rocksDB = rocksDB;
  }

  @PostMapping("/{key}")
  public ResponseEntity<String> save(@PathVariable("key") String key, @RequestBody String value) {
    log.info("RocksApi.save");
    rocksDB.save(key, value);
    return ResponseEntity.ok(value);
  }

  @GetMapping("/{key}")
  public ResponseEntity<String> find(@PathVariable("key") String key) {
    log.info("RocksApi.find");
    String result = rocksDB.find(key);
    if(result == null) return ResponseEntity.noContent().build();
    return ResponseEntity.ok(result);
  }

  @DeleteMapping("/{key}")
  public ResponseEntity<String> delete(@PathVariable("key") String key) {
    log.info("RocksApi.delete");
    rocksDB.delete(key);
    return ResponseEntity.ok(key);
  }

Now with the application ready you can build the project with maven mvn clean package and then run it mvn spring-boot:run, once the application starts you can test the basic operations using the API we just created using curl, the options used in the curl command below are to print headers and the payload from the requests so you can see a bit better what's going on:

Add entries: curl -v -H "Content-Type: text/plain" -X POST http://localhost:8080/api/rocks/1 -d mypersistedvalue, if you add a new entry with the same key(the 1 in the path) it will automatically override the previous one, notice the key can be any string in this simple implementation, here we're just using 1 as key for the example which is parsed as a String.

Get entries: curl -iv -X GET -H "Content-Type: text/plain" http://localhost:8080/api/rocks/1

Delete entries: curl -X DELETE http://localhost:8080/api/rocks/1

Done! This was a very simple initial reference meant to get you started, if you ever need to use a client-side persistent cache and decide RocksDB is a good option, this is surely just the tip of it, RocksDB gives you many other possible options when you need to customize your cache, you can set filters(like a BloomFilter) you can configure LRU caches and many other options are available, check out their full documentation and references on the Reference section below for further details.

Note: Kafka uses RocksDB by default when doing Streams and using KTables or GlobalKTables.

References

Clone the source code from github rocksdbBootApp : git clone git@github.com:stockgeeks/rocksdbBootApp.git

RocksJava documentation and project

Header Photo by Christopher Gower on Unsplash

Simplest Spring Kafka Producer and Consumer - Kotlin version

Marcos Maia — Mon, 07 Oct 2019 20:15:19 +0000

Let's now build and run the simplest example of a Kotlin Kafka Consumer and Producer using spring-kafka. If you need assistance with Kafka, spring boot or docker which are used in this article, or want to check out the sample application from this post please check the References section below, but for a quick access to the source code you can just: git clone git@github.com:stockgeeks/spirng-kafka-kotlin.git

These are the steps we're going to follow in this post:

Create a spring-boot Kotlin application, java 11 build with Gradle or Maven.
Create a Spring Kafka Kotlin Consumer.
Run local Kafka and Zookeeper using docker and docker-compose.
Produce some messages from the command line console-producer and check the consumer log.
Create a Spring Kafka Kotlin Producer.
Add some custom configuration.
Build an endpoint that we can pass in a message to be produced to Kafka.
Build and run the application with Maven or Gradle.
Send a message with curl and check the console messages to follow the message from the endpoint -> producer -> kafka -> consumer.

Create a Spring Boot Kotlin Application

Navigate to spring initializer, check this other post as an example if you want some more details, when creating the project pick Gradle OR Maven as build system and Kotlin, if you have never used spring initializer check this other post which covers it using Java(https://dev.to/thegroo/spring-boot-crash-course-21nm), make sure to have spring-kafka dependency added and select Java 11 under Options.

Create a Kotlin Spring Kafka Consumer

Let's now write the simplest possible Kotlin Kafka consumer with spring-kafka using spring-boot default configurations.

Create a class called KotlinConsumer:

import org.slf4j.LoggerFactory
import org.springframework.kafka.annotation.KafkaListener
import org.springframework.stereotype.Component

@Component
class KotlinConsumer {
  private val logger = LoggerFactory.getLogger(javaClass)
  @KafkaListener(topics = ["simple-message-topic"], groupId = "simple-kotlin-consumer")
  fun processMessage(message: String) {
    logger.info("got message: {}", message)
  }
}

This is it, this is all it takes because we are relying on spring-boot default configurations the key and value are configured to use be String by default.

Start Kafka and Zookeeper

As we've seen in details in this other article, we're going to be using docker-compose to run our local Kafka for development, let's start our Kafka and Zookeeper containers, if you didn't checkout the source code you can create a docker-compose.yml file in the root of your project and add the content from the link, let's start Kafka and Zookeeper:

docker-compose up -d

Make sure the containers are running:

docker ps

You should see Kafka and Zookeeper running:

Run the app

Let's now compile and run the application, if you need more detailed instructions please check this post, run the following commands to build and run the application:

using gradle(w):

gradle build && gradle bootRun

OR if you're using mvn(w):

mvn clean package && mvn spring-boot:run

The optional command with w in the end(mvnw or gradlew) applies if you don't have maven or gradle installed and prefers using the wrappers which are created by default by the start.spring.io site.

The application will start and you will see on the standard output the configurations for the consumer, the Kafka version being used and a message Started SpringKafkaApplication in x seconds.

Make sure to keep the application running, don't close the terminal window where it's running. Let's now produce a few messages with the Kafka console producer and see our consumer processing the messages and logging them out.

Produce message using the Kafka console producer

Open a new terminal and enter the Kafka running container so we can use the console producer:

docker exec -it kafka /bin/bash

Once inside the container cd /opt/kafka/bin , the command line scripts for Kafka in this specific image we're using are located in this folder. If you're using different docker images those scripts might be in some other location.

Run the console producer which will enable you to send messages to Kafka:

./kafka-console-producer.sh --broker-list localhost:9092 --topic simple-message-topic

The console will now block and you can write your message and hit enter, for each time you do this one message will be produced to the simple-topic. Try sending a few messages and watch the application standard output in the shell where you are running your Spring Boot application processing the messages and printing them.

Write a simple Spring Kotlin Producer

Time to create our spring-kafka producer. Create a class called KotlinProducer, we will again use the defaults for the producer as we did for the consumer.

import org.springframework.kafka.core.KafkaTemplate
import org.springframework.stereotype.Component

@Component
class KotlinProducer(private val kafkaTemplate: KafkaTemplate<String, String>) {

  fun send(message: String) {
    kafkaTemplate.send("simple-message-topic", message)
  }

}

That's it, all it takes as again we're using the Spring Boot defaults which assumes String for the Value of message and we're omitting the key as it's not mandatory, we will discuss how the key is used by Kafka in future articles.

Write an endpoint

Let's now create a simple endpoint which will receive a text message and publish it to Kafka, we're always returning 200 OK for now for simplicity of an example.

import org.springframework.web.bind.annotation.PostMapping
import org.springframework.web.bind.annotation.RequestBody
import org.springframework.web.bind.annotation.RequestMapping
import org.springframework.web.bind.annotation.RestController

@RestController
@RequestMapping("/api")
class MessageApi(private val kotlinProducer: KotlinProducer) {
  @PostMapping("/message")
  fun publish(@RequestBody message: String) {
    kotlinProducer.send(message)
  }
}

Again a simple piece of code relying on Spring Boot defaults and using it's annotations. We inject the producer and call it directly passing in any messages sent using the POST call.

Build and run the application.

using gradle:

gradle build && gradle bootRun

using maven:

mvn clean package && mvn spring-boot:run

There's a good chance you will get and error when running the application on your development machine now, this happens because your application is running inside your normal host network and Kafka and zookeeper are running inside the "docker network".

There are some ways to solve this, the best is to pass in your development machine hostname to docker-compose when starting the containers, if you open the docker-compose file from this project it has an entry at the KAFKA_ADVERTISED_LISTENERS: ... LISTENER_DOCKER_EXTERNAL like ${DOCKER_HOST_IP:-kafka}:9092 this tells compose to try to use the passed in hostname or Kafka by default, check the comments in the compose file to find out how to fix it and check the references section below for more details.

Send some messages with curl

Now make sure to watch the application terminal and in another terminal window let's use curl to send some messages:

curl -X POST http://localhost:8080/api/message -d "yet more fun" -H "Content-Type: text/plain"

You should see the response on the same terminal where curl was executed also check the consumer processing the message and printing it to the terminal where the application is running.

Done

This is it, it's done. You have now created the simplest possible Spring Boot application that produces and consume messages from Kafka using Kotlin. The reason it looks so simple is that we are relying on spring-boot and spring-Kafka default configurations as mentioned before.

If you want to know more about how Spring Boot or Kafka Works please take a look at the links in the next session where you'll find some references with further details.

If you have problems to connect your kafka client from your local development machine to the broker running in docker or Kubernetes please check the section Connecting a Kafka client on the end of this other post where you'll find details on how to fix it and links explaining why it happens and sort this out completely, have fun.

We're going to cover testing your consumer and producers in another post. Happy Coding.

References

Source code with the application created in this post.

To set up your environment with java, maven, docker and docker-compose, please check how to set up your environment for the example tutorials.

If you need some quick introduction to Kafka: Kafka - Crash Course

For some insights on how to use docker-compose for local development, please check this post: One to run them all where you will also learn some useful Kafka commands.

If you're new to Spring Boot, please take a look at Spring Boot - Crash Course

Docker compose environment variables to understand the configuration for the Kafka Advertise Listener.