DEV Community: Dan Newton

I rarely need Stack Overflow anymore

Dan Newton — Sun, 12 Jun 2022 00:00:00 +0000

I noticed recently that my dependence on Stack Overflow and similar platforms has wained as I have progressed in my career. The challenges I face and the questions arising from them have changed, becoming more abstract, handy-wavy, or super-specific to my company’s business logic. The likelihood that someone has already answered these sorts of questions becomes unlikely. Maybe, I could try to ask these questions myself (I’m more of a reader of Stack Overflow rather than a contributing member), but where would I even begin. These questions aren’t asked because of the amount of context and background knowledge required to comprehend the question.

Now, don’t get me wrong. I’m not bashing Stack Overflow at all here. There are plenty of questions that can be asked or are already answered, primarily revolving around using language features, libraries and frameworks. This is where it shines, which is why “rarely” instead of “do not” is included in the title. When I’m working on my own code outside of work, it is invaluable to me. But, that’s because my questions shift into insight about language features, libraries and frameworks like I mentioned a second ago.

At this point, you are probably wondering what point I’m trying to make, and believe me, I am also trying to figure that out right now. So here I go…

If, like me, you are trying to find answers to challenges you face at work, varying from writing better code, to solving architectural issues, to even the most abstract questions of “are we solving the correct problems”; point yourself in the direction of your colleagues. They (most of the time) have the context and foundational knowledge to correctly answer questions in a way that no one else can. I can’t even begin to quantify the amount I’ve learnt from turning to my teammates for help. They may have solved similar problems in the past and can share that experience with you, all based on their ability to actually understand the nuance in what you’re asking.

For more senior engineers reading this, do not underestimate the impact sharing your knowledge has on your teammates. Every time I hear the phrase, “At my old place”, I know we’re about to for a trip through their experiences, and I also know that without these conversations, I would not have progressed at the rate I have in my own career. You cannot rely on Stack Overflow and similar platforms to level up your team, it may help, but it will not produce well-rounded engineers.

For everyone else, you cannot rely on Stack Overflow and similar platforms for everything. Read longer-form content, such as blogs, books or podcasts, focusing more on the why than the how. However, having a back and forth between other engineers will be one of the most efficient ways to improve. If you fail to do this, you will never progress to the point where you rarely need Stack Overflow anymore.

Using GraphQL Java with Spring

Dan Newton — Mon, 10 Jan 2022 00:00:00 +0000

GraphQL Java is one of the most popular GraphQL server-side implementations for Java that I've found (with over 5k stars at the time of writing). If you're planning to expose a GraphQL API from a Java or JVM application, then this is an excellent library to start using.

This blog post will cover how to use GraphQL Java within a Spring application that exposes an endpoint for clients to send queries to. GraphQL Java does have its own official documentation covering this subject; however, I found it a bit too simplistic, which made it harder for me to wrap my head around it. I hope you don't think the same about the content of this post, although I guess you could write your own that is even more complicated than my examples!

I will be writing this from the assumption that you understand some of the basics of GraphQL. I won't be covering anything extremely complicated, so the basics will give you a good base for the content of this blog post. Knowing how to create a schema type and a query that does nothing fancy will be all you need. You can find this information from the official graphql.org site.

Also, here's a heads up that I have written my examples in Kotlin, although I tried to keep it friendly for Java readers.

Dependencies

Below are the Spring and GraphQL related dependencies used in this post:

<parent>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-parent</artifactId>
  <version>2.6.1</version>
</parent>

<dependencies>
  <dependency>
    <groupId>com.graphql-java</groupId>
    <artifactId>graphql-java</artifactId>
    <version>16.2</version>
  </dependency>
  <dependency>
    <groupId>com.graphql-java</groupId>
    <artifactId>graphql-java-spring-boot-starter-webmvc</artifactId>
    <version>2.0</version>
  </dependency>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
    <version>2.6.1</version>
  </dependency>
</dependencies>

The schema we'll use

Below is the GraphQL schema that we'll use throughout this post:

type Query {
  people: [Person]
  peopleByFirstName(firstName: String): [Person]
  personById(id: ID): Person
}

type Person {
  id: ID,
  firstName: String,
  lastName: String
  relationships: [Relationship]
}

type Relationship {
  relation: Person,
  relationship: String
}

I'll cover how we register and interact with the schema later on; knowing the shape of the types in the schema will set up the following sections.

Fetching data with DataFetchers

GraphQL Java uses DataFetchers to fetch data to include in the result of a query. More specifically, a DataFetcher retrieves the data for a single field when a query is executed.

Every field has an assigned DataFetcher. When an incoming GraphQL query is received, the library will call the registered DataFetcher for each field in the query.

The official DataFetcher documentation can be found here.

It is worth pointing out now, as it personally caused me a lot of confusion to begin with, that a "field" can mean two things:

The name of a query.
A property/field in a schema type.

This is important as it means that a DataFetcher can be linked to a query. In fact, every query must have an associated DataFetcher. Not doing this will cause GraphQL query requests to fail as there is no entry point to begin processing the query.

I keep mentioning DataFetchers, but what are they really in terms of code? Below is the DataFetcher interface:

public interface DataFetcher<T> {

  T get(DataFetchingEnvironment environment) throws Exception;
}

So when I say DataFetcher, I'm really talking about implementations of the DataFetcher interface.

Let's look at an example DataFetcher that will respond to the people query:

type Query {
    people: [Person]
}

The DataFetcher for this query:

@Component
class PeopleDataFetcher(private val personRepository: PersonRepository) : DataFetcher<List<PersonDTO>> {

  override fun get(environment: DataFetchingEnvironment): List<PersonDTO> {
    return personRepository.findAll().map { person -> PersonDTO(
        person.id,
        person.firstName,
        person.lastName
      ) 
    }
  }
}

The PeopleDataFetcher returns a List<PersonDTO> to correspond to the [Person] specified as the turn type of the GraphQL query. The PersonDTO contains the same fields found in the Person GraphQL type:

data class PersonDTO(val id: UUID, val firstName: String, val lastName: String)

A field for relationships has been omitted to simplify the examples

When PeopleDataFetcher.get is executed, it queries the database, maps the results to PersonDTOs and returns them.

When an incoming GraphQL query, like the following is received:

query {
    people {
        firstName
        lastName
        id
    }
}

The following is returned after GraphQL Java has called the PeopleDataFetcher:

{
  "data": {
    "people": [
      {
        "firstName": "John",
        "lastName": "Doe",
        "id": "00a0d4f2-637f-469c-9ecf-ba8839307996"
      },
      {
        "firstName": "Dan",
        "lastName": "Newton",
        "id": "27a08c14-d0ad-476c-ba09-9edad3e4c8f9"
      }
    ]
  }
}

That covers a first look at what DataFetchers do. We'll expand on them in the following sections to improve your understanding of GraphQL Java.

The default PropertyDataFatcher

As mentioned in the previous section, every field must have an assigned DataFetcher. That means for the Person type:

type Person {
  id: ID,
  firstName: String,
  lastName: String
  relationships: [Relationship]
}

You would need to associate DataFetchers to:

Person.id
Person.firstName
Person.lastName
Person.relationships

That seems a bit onerous, especially when you might be able to retrieve all of this data in one go. For example, retrieving data from a database could be the difference between 1 SQL query vs 4.

To resolve this, any field without an assigned DataFetcher uses the PropertyDataFetcher by default.

The PropertyDataFetcher uses various methods (e.g. getters or keys in a map) to extract field values from the parent field (could be a query or schema type).

To provide a concrete example, the PeopleDataFetcher we saw previously was used to respond to the people GraphQL query (the query, type and PeopleDataFetcher are included below):

type Query {
  people: [Person]
}

type Person {
  id: ID,
  firstName: String,
  lastName: String
  relationships: [Relationship]
}

@Component
class PeopleDataFetcher(private val personRepository: PersonRepository) : DataFetcher<List<PersonDTO>> {

  override fun get(environment: DataFetchingEnvironment): List<PersonDTO> {
    return personRepository.findAll().map { person -> PersonDTO(
        person.id,
        person.firstName,
        person.lastName
      ) 
    }
  }
}

The PeopleDataFetcher returns a PersonDTO for each Person it finds to respond to the top-level query. This can be thought of as the "parent field".

The GraphQL library will then move down to fetch values for each queried field in Person, for example, firstName and lastName. Using the PropertyDataFetcher, it accesses each PersonDTO returned by the parent field's DataFetcher (PeopleDataFetcher) and extracts the values using their getters.

In concrete terms, this means:

Person.id -> Value provided by PersonDTO.id.
Person.firstName -> Value provided by PersonDTO.firstName.
Person.lastName -> Value provided by PersonDTO.lastName.
Person.relationships -> Empty because no value was provided.

PersonDTO.id is how you write Person.getId in Kotlin (the applies to the other fields).

You might need to run that through your head a few times so it makes sense. I only properly understood this after debugging the library when my code wasn't working correctly.

Writing a DataFetcher for a schema type's field

The PeopleDataFetcher we've seen throughout this post responds to a query. Now let's look at a custom DataFetcher that should be associated with a schema type's field.

The PersonRelationshipsDataFetcher fetches data for the Person.relationships field:

@Component
class PersonRelationshipsDataFetcher(
  private val relationshipRepository: RelationshipRepository
) : DataFetcher<List<RelationshipDTO>> {

  override fun get(environment: DataFetchingEnvironment): List<RelationshipDTO> {
    // Gets the object wrapping the [relationships] field
    // In this case a [PersonDTO] object.
    val source = environment.getSource<PersonDTO>()
    return relationshipRepository.findAllByPersonId(source.id).map { relationship ->
      RelationshipDTO(
        relation = relationship.relatedPerson.toDTO(),
        relationship = relationship.relationship
      )
    }
  }
}

It looks similar to the PeopleDataFetcher we saw previously, except for the new call to DataFetchingEnvironment.getSource. This method allows a DataFetcher to access the object returned by the DataFetcher associated with the parent field. After accessing this object, information is extracted from it (PersonDTO.id) to be used in the SQL query executed by the PersonRelationshipsDataFetcher.

Writing a DataFetcher for a query containing an argument

Queries become far more worthwhile when you can pass arguments into them.

Take the query:

type Query {
  peopleByFirstName(firstName: String): [Person]
}

To handle this you'd want a DataFetcher like the one below:

@Component
class PeopleByFirstNameDataFetcher(private val personRepository: PersonRepository) : DataFetcher<List<PersonDTO>> {

  override fun get(environment: DataFetchingEnvironment): List<PersonDTO> {
    // The argument is extracted from the GraphQL query
    val firstName = environment.getArgument<String>("firstName")
    return personRepository.findAllByFirstName(firstName)
      .map { person -> PersonDTO(person.id, person.firstName, person.lastName) }
  }
}

The important method call here is to DataFetchingEnvironment.getArgument, which does as it says and extracts an argument from the incoming GraphQL query. Handily, getArgument allows you to specify the type the argument should be (so you don't have to convert it yourself).

DataFetchingEnvironment also contains other methods regarding arguments, e.g. getArguments and containsArgument.

Setting up a GraphQL instance

You've seen how to write a few DataFetchers by this point; we now need to tie everything together by creating a GraphQL instance and registering an application's DataFetchers.

The @Configuration code below does just that:

@Configuration
class GraphQLConfiguration(
  private val peopleByFirstNameDataFetcher: PeopleByFirstNameDataFetcher,
  private val peopleDataFetcher: PeopleDataFetcher,
  private val personByIdDataFetcher: PersonByIdDataFetcher,
  private val personRelationshipsDataFetcher: PersonRelationshipsDataFetcher
) {

  @Bean
  fun graphQL(): GraphQL {
    val typeRegistry: TypeDefinitionRegistry = SchemaParser().parse(readSchema())
    val runtimeWiring: RuntimeWiring = buildWiring()
    val graphQLSchema: GraphQLSchema =  SchemaGenerator().makeExecutableSchema(typeRegistry, runtimeWiring)
    return GraphQL.newGraphQL(graphQLSchema).build()
  }

  private fun schemaFile(): File {
    return this::class.java.classLoader.getResource("schema.graphqls")
      ?.let { url -> File(url.toURI()) }
      ?: throw IllegalStateException("The resource does not exist")
  }

  private fun buildWiring(): RuntimeWiring {
    return RuntimeWiring.newRuntimeWiring()
      .type(newTypeWiring("Query").dataFetcher("peopleByFirstName", peopleByFirstNameDataFetcher))
      .type(newTypeWiring("Query").dataFetcher("people", peopleDataFetcher))
      .type(newTypeWiring("Query").dataFetcher("personById", personByIdDataFetcher))
      .type(newTypeWiring("Person").dataFetcher("relationships", personRelationshipsDataFetcher))
      .build()
  }
}

The purpose of this @Configuration class is to create a GraphQL instance that GraphQL Java uses. Further setup is not required as it will be picked up by Spring Boot's auto-configuration.

The first step of creating the GraphQL instance requires reading the application's GraphQL schema. SchemaParser.parse can take in Files, InputStreams, Readers or Strings, which it parses (as the class name suggests) and returns a TypeDefinitionRegistry to be used later. In this application, the schema is defined in a resource file which gets fed into SchemaParser.parse. This is what allows the GraphQL library to understand incoming queries and what can or cannot be handled.

The DataFetchers are then registered with a RuntimeWiring instance (through a RuntimeWiring.Builder returned by RuntimeWiring.newRuntimeWiring). Every time I mentioned "gets the DataFetcher associated to the field", this is where the association actually happens. I can stop hand waving all the time now since you've seen the code.

Each DataFetcher in this example application is injected into the configuration class and linked to the RuntimeWiring instance through its type method. Each TypeRuntimeWiring.Builder instance (returned by newTypeWiring) requires 3 fundamental inputs:

The name of the schema type ("Query" or a type name).
The name of the field (query name or schema type field).
The DataFetcher associated with the type and field.

After registering each DataFetcher, the RuntimeWiring instance is finalised using build.

Finally, the TypeDefinitionRegistry and RuntimeWiring created previously are passed through a SchemaGenerator and then into GraphQL.newGraphQL to retrieve a fully functional GraphQL instance.

Sending a GraphQL query to the application

With the setup complete, the application now exposes a /graphql endpoint provided by the auto-configured code in graphql-java-spring-boot-starter-webmvc. This endpoint is where clients will send GraphQL queries to.

In this section, we'll look at how to send a query using cURL and Postman (which has GraphQL functionality) and view the returned data.

The query we are trying to send:

query {
    peopleByFirstName(firstName: "Dan") {
        firstName
        lastName
        id
        relationships {
            relation {
                firstName
                lastName
            }
            relationship
        }
    }
}

cURL:

  curl 'localhost:8080/graphql/' \
  -X POST \
  -H 'content-type: application/json' \
  --data '{ "query": "query { peopleByFirstName(firstName: \"Dan\") { firstName lastName id relationships { relation { firstName lastName } relationship }}}"}'

Postman:

Both of these methods return the same data (I would be worried if they didn't):

{
  "data": {
    "peopleByFirstName": [
      {
        "firstName": "Dan",
        "lastName": "Newton",
        "id": "27a08c14-d0ad-476c-ba09-9edad3e4c8f9",
        "relationships": [
          {
            "relation": {
              "firstName": "Laura",
              "lastName": "So"
            },
            "relationship": "Wife"
          },
          {
            "relation": {
              "firstName": "Random",
              "lastName": "Person"
            },
            "relationship": "Friend"
          }
        ]
      },
      {
        "firstName": "Dan",
        "lastName": "Doe",
        "id": "3c07b717-8b9c-4d88-926f-c892be38ee85",
        "relationships": []
      },
    ]
  }
}

The most crucial factor here is that a POST request is used. It seems standard for GraphQL APIs to use POST requests for both fetching and mutating data. This threw me off for a while, as the error I received from the /graphql endpoint didn't cause me to believe it was due to using the wrong HTTP verb.

For clarity, using the wrong HTTP verb (e.g. a GET) through Postman leads to the following response and logline:

{
    "timestamp": "2022-01-03T16:50:58.376+00:00",
    "status": 400,
    "error": "Bad Request",
    "path": "/graphql"
}

Resolved [org.springframework.web.bind.MissingServletRequestParameterException: Required request parameter 'query' for method parameter type String is not present]

Improving the registration of DataFetchers

The registration of DataFetchers could be improved as they are currently registered by injecting every DataFetcher by name into the GraphQLConfiguration class and manually associating each with a type and field name. This isn't too bad right now since the application is small; if you argued that it already looks dodgy, then I'd agree with you.

To make the code more maintainable and extensible moving forward, we can introduce a new structure for defining DataFetchers and registering them.

We can achieve this by defining a new interface that specifies the type and field name a DataFetcher should be linked to:

/**
 * [TypedDataFetcher] is an instance of a [DataFetcher] that specifies the schema type 
 * and field it processes.
 *
 * Instances of [TypedDataFetcher] are registered into an instance of [RuntimeWiring] 
 * after being picked up by Spring (the instances must be annotated with @[Component]
 * or a similar annotated to be injected).
 */
interface TypedDataFetcher<T> : DataFetcher<T> {

  /**
   * The type that the [TypedDataFetcher] handles.
   *
   * Use `Query` if the [TypedDataFetcher] responds to incoming queries.
   *
   * Use a schema type name if the [TypedDataFetcher] fetches data for a single field
   * in the specified type.
   */
  val typeName: String

  /**
   * The field that the [TypedDataFetcher] should apply to.
   *
   * If the [typeName] is `Query`, then [fieldName] will be the name of the query the 
   * TypedDataFetcher] handles.
   *
   * If the [typeName] is a schema type, then [fieldName] should be the name of a single
   * field in [typeName].
   */
  val fieldName: String
}

By using TypedDataFetcher, you can retrieve all its implementations as Spring allows injecting of Lists containing all instances implementing an interface.

Combining this interface with modifications to the DataFetcher registration code in GraphQLConfiguration brings it all together:

@Configuration
class GraphQLConfiguration(private val dataFetchers: List<TypedDataFetcher<*>>) {

  // Create the GraphQL instance (no changes from the previous example).

  /**
   * Loops through all injected [TypedDataFetcher] instances and includes them in the output [RuntimeWiring] instance.
   */
  private fun buildWiring(): RuntimeWiring {
    val wiring = RuntimeWiring.newRuntimeWiring()
    for (dataFetcher in dataFetchers) {
      wiring.type(newTypeWiring(dataFetcher.typeName).dataFetcher(dataFetcher.fieldName, dataFetcher))
    }
    return wiring.build()
  }
}

The registration now uses the typeName and fieldName provided by each TypedDataFetcher, breaking the rigid link between DataFetcher implementations and the types, fields or queries they represent. Adding new TypedDataFetchers becomes straightforward after making this alteration; you create a new implementation, define the typeName and fieldName, annotate with it @Component, and the rest is handled for you.

Summary

GraphQL Java lets you do what the name suggests, support GraphQL queries within Java (or other JVM languages).

It does this through associating DataFetchers to types and fields that you write and pointing incoming queries their way. With the library handling the parsing of queries, it frees you to focus on implementing the DataFetchers that contain your application's primary functionality.

I'd love to try and summarise how to use GraphQL Java further; however, I already struggled to write the post in the most coherent way as possible; there is no way I could condense it to a few sentences.

Anyway, the end!

Intro to Kafka - Ordering related records

Dan Newton — Fri, 06 Aug 2021 00:00:00 +0000

Ordering records in Kafka requires you to make a conscious decision to construct your system in a way that supports ordering. You don’t have to do this, as it depends on the functionality of your system. You don’t need ordering for all situations, however, many problems require ordering between events (or records in Kafka) to safely and adequately implement a process.

So how can records be ordered?

We can answer the question by using our knowledge about topics, partitions and how consumers retrieve records. The important information we need is as follows:

Consumers read records from partitions inside topics.
Each new record is inserted with an incremented offset based on the previous maximum offset.
These records are read from the lowest to highest offset values.
Records with the same key are always stored in the same partition.

By combining this information, we can infer that ordering between records can be maintained when they have the same key. By ensuring that all records that have some relationship between them have the same key, you can guarantee that they will be placed into the same partition, where a consumer will now retrieve them in the order they were stored.

I have also put together a diagram summarising the information above:

If you think about this from a more application-level perspective and replace “related records” with something with more context, say, a customer identifier, it becomes easier to understand how you could leverage the ordering provided by Kafka.

All records for a customer would have their keys set to the customer’s identifier, which guarantees Kafka will order any events that this customer triggered.

Code-wise, this might look like:

fun pay(customerId: UUID, amount: BigDecimal) {
  producer.send(
    ProducerRecord(
      PAYMENTS, // topic
      customerId, // key
      amount // amount
    )
  )
}

The customerId is a value that represents a single customer. By using this value as the key for all records related to a single customer, Kafka will place the records into the same partition. When producers read from the PAYMENTS topic, they will receive records in the order they were created, making writing the downstream processing logic more straightforward.

Now, I’m going to chuck a bit more information at you, that you don’t need to know. However, it will fill in the gaps you might already be questioning on how it works.

More precisely, how does a topic decide which partition to store a record in?

A formula determines the a record’s partition:

hash($key) % number_of_partitions

Where:

hash A function that hashes the input $key with Kafka’s chosen digest algorithm.
$key The key of a record.
number_of_partitions - The number of partitions in the topic.

% (Mod/Modulo) returns the remainder when dividing the left side by the right (e.g. 20 % 7 = 6)

Below is a diagram that visualises the process of determining the partition of a record:

Note, this is assuming that the default formula for calculating a record’s partition is used. You can configure this if desired, more information can be found in the Kafka documentation and in their JavaDocs (although the amount of information is honestly a bit lacking).

Summary

Records within a single partition are ordered by insertion order.
Records with the same key are stored in the same partition (when using default settings).
This ensures ordering between records with the same key.
The formula for calculating a record’s partition is hash($key) % number_of_partitions.

Intro to Kafka - Consumer groups

Dan Newton — Sun, 20 Jun 2021 00:00:00 +0000

Consumers can form groups, aptly named “consumer groups”. These consumer groups determine what records a consumer receives. I know that sounds quite vague, but honestly, I can’t think of any other short and snappy way to summarise this topic.

To be more specific, consumer groups do the following:

Groups consumers by function
Shares a topic’s partitions between consumers in the same group

The following sections expand on these points.

Groups consumers by function

Consumer groups allow you to relate a set of consumers working together to perform a single function, process or task. Every consumer group will process every record in a topic independently from other groups.

For example, suppose you have a topic representing sales data. In that case, one group could maintain an aggregate of the sale values while another is passing it to a downstream system for later processing. Both of these groups will run independently and manage their offsets, even when subscribing to the same topic.

The diagram below has two consumer groups who receive all the records stored in the topic:

Shares a topic’s partitions between consumers in the same group

Every consumer inside a consumer group is assigned a number of a topic’s partitions, ensuring that records are shared between the group while also keeping each partition’s records in order. This mechanism is made more efficient by keeping the assigned partitions the same until introducing a new consumer to the group. At this point, Kafka reassigns the partitions between the group’s consumers, and this process is called “rebalancing”. Consumers can also be assigned multiple partitions if there are more partitions than consumers within a group.

The diagram shows the sharing of partitions within a single consumer group:

Introducing another consumer group does not affect the assignment of existing groups, as shown below:

The system will rebalance when a new consumer is added to one of the groups, visualised below:

A new consumer was added to group 1, which caused the assigned partitions to rebalance between them.

Creating a consumer group

You don’t have to create a group explicitly. Instead, it’s done as part of creating a consumer.

When constructing a consumer, pass in the group.id property of the group. In fact, this field is required, so any Kafka example or tutorial you’ve followed will have already included this piece of code.

fun createConsumer(): Consumer<String, String> {
  val props = Properties()
  props["bootstrap.servers"] = "localhost:9092"
  // The important stuff is below!
  props["group.id"] = "test"
  // That's the end of the important stuff!
  props["enable.auto.commit"] = "true"
  props["auto.commit.interval.ms"] = "1000"
  props["key.deserializer"] = "org.apache.kafka.common.serialization.StringDeserializer"
  props["value.deserializer"] = "org.apache.kafka.common.serialization.StringDeserializer"
  return KafkaConsumer(props)
}

Initialising more consumers adds them to the existing group without you having to do anything else. Kafka makes this work exceptionally smoothly since the Broker knows about all the consumers in the group and can therefore handle consumers joining (or leaving) a group.

How partitions are assigned

The chosen “assignment strategy” determines the assignment of partitions within a group.

The default strategy is the RangeAssignor, which I would rather not explain in my own words, so I’ll quote Kafka’s JavaDocs instead.

The range assignor works on a per-topic basis. For each topic, we lay out the available partitions in numeric order and the consumers in lexicographic order. We then divide the number of partitions by the total number of consumers to determine the number of partitions to assign to each consumer. If it does not evenly divide, then the first few consumers will have one extra partition.

For example, suppose there are two consumers C0 and C1, two topics t0 and t1, and each topic has 3 partitions, resulting in partitions t0p0, t0p1, t0p2, t1p0, t1p1, and t1p2.

The assignment will be:

C0: [t0p0, t0p1, t1p0, t1p1] C1: [t0p2, t1p2]

You can configure a consumer group’s assignment strategy by using the partition.assignment.strategy property and defining the qualified class name of the assignor to use.

More information on the different assignors can be found in Kafka’s documentation.

Summary

Consumer groups in Kafka allow you to:

Group consumers by their function in a system.
Split the processing load of a topic by sharing its partitions between consumers in a group.

You’ve seen how to create a consumer and place it in a group or add one to an existing group by specifying its group.id during initialisation. You will hopefully also recall that Kafka does the rest of the work and that selecting the group.id is all you have to do.

By finishing this post, we have now covered the core parts of Kafka (at least in my opinion), and you should have enough knowledge to start building applications using it. Probably nothing fancy, but you’ll at least have the foundational understanding to get started.

I’ll look at particular topics more closely in future posts that I might have brushed over previously, or I’ll cover new material as I learn it myself.

Intro to Kafka - Consumers

Dan Newton — Sun, 09 May 2021 00:00:00 +0000

Consumers make up the Kafka client code that reads records from topics, working at the opposite end of the chain as producers and benefiting from the same client features such as connection pooling and network buffering. Thus, providing you with a concise API that still allows you to write highly effective applications.

The diagram below visualises the interaction between a consumer and broker, where the consumer receives records stored in a topic.

In the following sections of this post, we will take a closer look at how consumers work and the client code needed to leverage them in your applications.

The code written in this post is in Kotlin, but similar code should be applicable for any JVM language. Furthermore, clients in other languages will also look virtually the same.

Creating a consumer

The first step is creating a KafkaConsumer:

fun createConsumer(): Consumer<String, String> {
  val props = Properties()
  props["bootstrap.servers"] = "localhost:9092"
  props["group.id"] = "test"
  props["enable.auto.commit"] = "true"
  props["auto.commit.interval.ms"] = "1000"
  props["key.deserializer"] = "org.apache.kafka.common.serialization.StringDeserializer"
  props["value.deserializer"] = "org.apache.kafka.common.serialization.StringDeserializer"
  return KafkaConsumer(props)
}

When creating a KafkaConsumer, you must provide its configuration via a Properties object or a Map<String, Object>. In addition, KafkaConsumer’s constructor also accepts values for the key.deserializer and value.deserializer properties.

You can find the available configuration options in Kafka’s documentation.

Now that you have an initialised consumer, you’ll need to subscribe to some topics to receive records.

Subscribing to topics

A consumer must subscribe to topics to retrieve the records stored in them. To be clear, a single consumer can subscribe to multiple topics at once. You do not have to make a separate consumer per topic.

A consumer subscribes to topics using its subscribe method (seems pretty obvious, I know…).

consumer.subscribe(listOf("string name of topic"))

Even though subscribe sounds straight forward, it does quite a lot of work behind the scenes, dynamically assigning a consumer the partitions it receives records from. The assignment of partitions becomes essential when you start writing more exciting applications that contain a more significant number of consumers.

Polling (retrieving) records

After a consumer subscribes to some topics, it needs to request records to pass them off to the rest of the application. This process is called polling and is done using the poll function.

When calling poll, you must pass in a timeout duration to allow your application to move along if no records are available for retrieval.

ConsumerRecords<K, V> poll(Duration timeout);

When no records are passed to a consumer within the timeout duration, it will progress with an empty collection of records.

The objects returned from poll are ConsumerRecords. These are similar but different from the ProducerRecords used to create records stored within a topic.

ConsumerRecords have several extra fields and a few that become required compared to ProducerRecords.

A ConsumerRecord contains the following fields:

Topic - The topic to the record was stored in.
Partition - The partition to the record was stored in.
Timestamp - The timestamp of the record.
Key (optional) - The key of the record.
Value (optional) - The value of the record.
Headers - Extra metadata to go alongside the record’s value.
Offset - The offset that the record holds within its partition.
Timestamp type - Indicates how the record’s timestamp value was generated.
Serialized key size - The size of the serialized, uncompressed key in bytes. If key is null, the returned size is -1.
Serialized value size - The size of the serialized, uncompressed value in bytes. If key is null, the returned size is -1.
Leader epoch - The current value of the leader epoch (this StackOverflow post seems to give some information on how this value is generated).

You don’t have to worry about generating these fields as you will only be receiving them after calling poll. The values are either defined by a ProducerRecord or generated by Kafka.

Code-wise, polling records is done as follows:

val records: ConsumerRecords<String, String> = consumer.poll(Duration.ofSeconds(5))
for (record: ConsumerRecord<String, String> in records) {
  println("Consumer reading record: ${record.value()}")
}

ConsumerRecords implements Iterable and can therefore be iterated over directly.

poll returns a ConsumerRecords object which provides some convenience functions and can also be iterated over directly. For example, it allows you to extract all the records for a particular topic, which is handy since you can subscribe to multiple topics in a single consumer.

You’ll then want to loop over the returned ConsumerRecords and process them in whatever way your application desires.

Offsets

Every record stored in a partition is assigned an offset which denotes how far they are from the start of the partition. Adding new records increases the offset, guaranteeing that more recent records will always have a higher offset than older records (based on time stored).

Offsets are not significant when persisting new records as this part is all handled by Kafka; however, the concept plays an important role when consuming records.

Consumers use partition offsets to keep track of what records have been processed. As it polls records and computes whatever needs computing, it needs to note down the offsets of these records, so that the consumer knows what point to start polling records from in case of failure.

Committing offsets

The process of keeping track of a consumer’s offset is known as “committing”. In that it commits the current offset for a partition to storage.

A consumer does not store its offsets locally. Instead, it sends them to the Kafka broker, who then persists them to disk.

Maintaining offsets in this way keeps the importance of consumer clients low within the context of the whole application. Starting, stopping and even destroying the machines running the consumers is possible without preventing the application from recovering. You can initialise a new consumer client and instantly continue processing from where the previous consumer was.

Furthermore, it allows you to focus your efforts on securing the broker’s machines rather than maintaining maximum security on many consumer clients. The broker contains the critical information, and keeping offsets in this way works to ensure this.

To commit a consumer’s offsets to the Kafka broker, you should use one of the following methods:

// Consumer.java

// Synchronous offset committing
void commitSync();
void commitSync(Duration timeout);
void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets);
void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets, final Duration timeout);
// Asynchronous offset commtting
void commitAsync();
void commitAsync(OffsetCommitCallback callback);
void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback);

These methods all commit a consumer’s offsets and come with different degrees of control. For example, the overloads of commitSync and commitAsync that have no arguments are fully managed solutions that determine the offsets from the polled records — setting the offsets of each partition to the highest offset value contained in the received records.

The other options give you more control but require you to do more work and set the offsets yourself.

You will want to commit your offsets after processing records. This will form part of the standard process when working with Kafka. Poll records, process them and commit the updated offsets to confirm that they should not be received again.

The code for this process will look like:

val records: ConsumerRecords<String, String> = consumer.poll(Duration.ofSeconds(5))
for (record: ConsumerRecord<String, String> in records) {
  println("Consumer reading record: ${record.value()}")
}
consumer.commitAsync()

Committing can also be done periodically in the background if configured to do so, see the Kafka documentation - enable.auto.commit.

Polling loop

To finish the process of polling records, processing them and committing the updated offsets, you’ll want to group these steps and repeat them as your application requires. For most use cases, repeating these steps in a loop might suffice, which would provide a consumer that infinitely pulls records from a Kafka broker and feeds them to the rest of the application.

This loop resembles what you’ll see in the Consumer’s JavaDocs as well as many examples online.

The most straightforward way to do something infinitely is to use a while(true) loop run on a dedicated thread:

thread {
  while (true) {
    val records: ConsumerRecords<String, String> = consumer.poll(Duration.ofSeconds(5))
      for (record: ConsumerRecord<String, String> in records) {
        println("Consumer reading record: ${record.value()}")
      }
    consumer.commitAsync()
  }
}

I don’t have anything left to say here, so let’s close up this post.

Summary

The overall process of using a Kafka consumer consists of:

Creating a consumer.
Subscribing to some topics.
Polling records from the Kafka broker.
Processing these records with the rest of your application.
Committing the updated offsets back to the Kafka broker.
Repeat, by going back to step 3 (polling records).

I have also put together a nice diagram to showcase this (I think it’s nice anyway!).

Bringing all these steps together gives you the following code:

fun main() {
  val consumer = createConsumer()
  thread { consumeMessages(consumer, topic) }
}

fun createConsumer(): Consumer<String, String> {
  val props = Properties()
  props["bootstrap.servers"] = "localhost:9092"
  props["group.id"] = "test"
  props["enable.auto.commit"] = "true"
  props["auto.commit.interval.ms"] = "1000"
  props["key.deserializer"] = "org.apache.kafka.common.serialization.StringDeserializer"
  props["value.deserializer"] = "org.apache.kafka.common.serialization.StringDeserializer"
  return KafkaConsumer(props)
}

fun consumeRecords(consumer: Consumer<String, String>, topic: String) {
  consumer.subscribe(listOf(topic))
  while (true) {
    val records: ConsumerRecords<String, String> = consumer.poll(Duration.ofSeconds(5))
      for (record: ConsumerRecord<String, String> in records) {
        println("Consumer reading record: ${record.value()}")
      }
    consumer.commitAsync()
  }
}

We have now looked at how to write the consuming side of an application that leverages Kafka. Combining this along with the information covered in previous posts about topics, partitions and producers, you should now have enough knowledge to write applications that interact with Kafka.

In the following posts, we will cover topics (not a Kafka topic) that expand on what we covered so far to allow you to start writing more effective Kafka applications.

Intro to Kafka - Producers

Dan Newton — Sat, 24 Apr 2021 00:00:00 +0000

Producers make up the Kafka client code that sends records to topics. These clients dramatically reduce the complexity of leveraging Kafka as it handles problems like connecting pooling and network buffering. You can choose to interact with a small amount of API while still writing highly effective applications.

The diagram below demonstrates how a producer sends a record to a broker, who then records it in one of its topics’ partitions.

From a high level, this is how a producer interacts within a Kafka system. The following sections will explore how to use the producer API and the code required to do so.

The code written in this post is in Kotlin, but similar code should be applicable for any JVM language. Furthermore, clients in other languages will also look virtually the same.

Creating a producer

Create a KafkaProducer to start sending records to Kafka:

fun createProducer(): Producer<String, String> {
  val props = Properties()
  props["bootstrap.servers"] = "localhost:9092"
  props["acks"] = "all"
  props["retries"] = 0
  props["linger.ms"] = 1
  props["key.serializer"] = "org.apache.kafka.common.serialization.StringSerializer"
  props["value.serializer"] = "org.apache.kafka.common.serialization.StringSerializer"

  return KafkaProducer(props)
}

When creating a KafkaProducer, you must provide its configuration via a Properties object or a Map<String, Object>. KafkaProducer’s constructor also accepts values for the key.serializer and value.serializer properties.

You can find the available configuration options in Kafka’s documentation.

You’re now ready to start producing records, but first, you need to know what a record contains.

Creating records

From the simplest perspective, a record consists of a key, value, and the topic that stores the record.

However, there is much more than just this. In fact, neither the key nor value are actually required (in that they can be null).

This is what a record really looks like:

Ordered in the same order as required by ProducerRecord’s full constructor.

To reiterate, and because not everyone likes diagrams, a record consists of the following properties (along with some explanations):

Topic - The topic to send the record to.
Partition (optional) - The partition to send the record to. When not provided, the partition is calculated by hashing the record’s key and modding it against the number of partitions.
Timestamp (optional) - The timestamp of the record. When not explicitly provided, it will default to System.currentTimeMillis.
Key (optional) - The key of the record.
Value (optional) - The value of the record.
Headers (optional) - Extra metadata to go alongside the record’s value.

Yes, both the key and value can be null. However, when you think about it, it doesn’t seem too crazy. It’s perfectly valid to receive null objects from the rest of your application code and Kafka being ok with that likely saves you a headache further down the line. However, it is probably still worth thinking about null values to be a good little developer.

The code for creating a ProducerRecord is shown below:

val record: ProducerRecord<String, String> = ProducerRecord(
  topic, // Topic
  time.toString(), // Key
  "Record sent at $time" // Value
)

Nothing fancy to show here since I only include the topic, key and value. Several constructors help you create ProducerRecords; however, you might be forced to pass in some nulls to make the records your applications require.

Sending records

After creating a record to send, all that remains is actually sending it. The code required for this is straightforward:

producer.send(record)

send asynchronously sends a record to the Kafka broker, returning a Future that you can use to make the call blocking if you desire (by calling Future.get).

You can also register a callback that executes when the sending of the record is acknowledged:

producer.send(record) { metadata: RecordMetadata, exception: Exception ->
  log.info("Stored in topic: ${metadata.topic()}, partition: ${metadata.partition()} with offset: ${metadata.offset()}")
}

That’s it. The record provides the rest of the information, allowing the process of sending a record to remain simple.

Multiple producers

Kafka supports multiple producers without significantly affecting your applications.

You’ll need to create extra producers using the same process as starting a single producer. You don’t have to do anything special here.

Summary

The overall process for sending a record to a Kafka topic consists of:

Creating a producer.
Creating a record.
Sending the record via the producer.

The diagram below brings together all the pieces discussed in this post:

Bringing the 3 steps together gives you code similar to the following:

// Create a producer
fun createProducer(): Producer<String, String> {
  val props = Properties()
  props["bootstrap.servers"] = "localhost:9092"
  props["acks"] = "all"
  props["retries"] = 0
  props["linger.ms"] = 1
  props["key.serializer"] = "org.apache.kafka.common.serialization.StringSerializer"
  props["value.serializer"] = "org.apache.kafka.common.serialization.StringSerializer"

  return KafkaProducer(props)
}

fun produceMessages(producer: Producer<String, String>, topic: String) {
  // Execute every 2 seconds
  fixedRateTimer(daemon = true, period = Duration.ofSeconds(2).toMillis()) {
    val time = LocalDateTime.now()
    // Create a record
    val record: ProducerRecord<String, String> = ProducerRecord(
      topic, // Topic
      time.toString(), // Key
      "Record sent at $time" // Value
    )
    println("Producer sending message: $record")
    // Send the record
    producer.send(record)
  }
}

You now know how to formulate the producing side of an application that leverages Kafka. Next up is understanding how to use the created records on the other end of your application. I will cover consumers in my following post.

Intro to Kafka - Topics and partitions

Dan Newton — Mon, 12 Apr 2021 00:00:00 +0000

Topics

Kafka, at a high level, stores records in topics. When handing a new record off to Kafka, you must decide the name of the topic you wish to send it to. On the flip side, you must also provide a topic’s name to receive and process records out of Kafka.

Records, events, messages or whatever you want to call them equate to the same thing. Kafka seems to call them records, so that is what I will stick with within this post.

Think of topics as the queue or table that you want your data to be stored in. I quite liked the comparison between topics and database tables made in this talk by Robin Moffatt. For example, you have a topic that contains orders, which is quite reasonable to compare to a database table containing order information. Your applications can then use this topic to submit information about orders as they arrive while downstream services process them.

In Kafka terminology, producers submit new records, while consumers read these same records later (albeit possibly almost instantly). Topics tie producers and consumers together while creating a clear cut boundary between them.

Topics are denoted by name, meaning that as records are produced, they will include the name of the topic they should be stored in, while consumers can subscribe to several topics, again decided by name.

For example, this might look like:

The two diagrams above show what topics are in Kafka, but only from a high level. The following section will take a closer look at what goes on inside a topic.

Partitions

A single topic can consist of multiple internal queues, named partitions, which hold the records published to Kafka.

Each partition within a topic maintains its own order. Leveraging this fact is what allows applications that include Kafka to scale well (if done correctly).

Producers push records to any partition within a topic. When submitting a record, a partition is chosen, determined by its key, or explicitly specifying a partition. Divvying out records in this manner, allows each partition to grow at similar rates.

When consumers subscribe to a topic, they are assigned some partitions that they receive records from. For each partition, a consumer will receive records that maintain their produced order. Thanks to this ordering, you can start using multiple consumers to process events from a single topic to improve performance and availability.

The following diagram below gives a visual representation of how producers and consumers interact with a topic’s partitions.

Further explanations of partitions are complicated without looking more into other concepts such as producers, consumers and consumer groups. I will cover all of these subjects, illuminating how every component of Kafka works together to build scalable, performant and resilient applications.

What The Mandalorian (season 2) teaches us about Software Engineering

Dan Newton — Wed, 23 Dec 2020 09:42:36 +0000

Season 2 of The Mandalorian isn't just a great TV programme that is exciting from start to finish. It is also packed with knowledge and lessons that, if you squint hard enough and really open your mind, can be applied to Software Engineering (and maybe even life in general).

There are 8 episodes in season 2, so here are 8 nuggets of wisdom for you. I chose not to do 1 per episode, because, well, I got stuck. Feel free to give it a try yourself if you want though!

Do note, this post is full of images that won't make sense if you haven't watched the episodes. Obviously, there are spoilers riddled throughout this post, so it's your choice if you continue or not.

Even a well thought out plan can fail. Sometimes, you need to take the general ideas and run with it.
Don't rush into a solution because part of it looks nice. Make sure you consider long term use and beware of bugs that could arise from your decisions.
Help juniors as much as you can, even if it makes your life harder a bit harder.
Give credit when it's deserved.
Don't always interrupt others when they are trying to concentrate.
A well-working team can tackle extremely challenging problems.
Allow your teammates, mentors and senior engineers to help you when you're in a jam.
Remember to show you face every now and then, especially if you're working from home.

If you have your own lessons learnt from this season, let me know in the comments!

Images were taken from:

Cover image - https://filmdaily.co/news/mandalorian-season-2-timeline/

Building a REST API with Quarkus

Dan Newton — Tue, 22 Sep 2020 00:00:00 +0000

Quarkus is a Java framework designed to run within containers. Focusing on fast start-up times and low memory usage making it more suitable to run within container orchestration platforms like Kubernetes.

I wanted to write about Quarkus for a while now, and have finally found the time to do so.

The content of this post will focus on writing a simple REST API using Quarkus. I will not explain how Quarkus works, because, well, I haven't put the time in yet to do so correctly. Hopefully, I can rectify this in future posts. Therefore, if you want more information about Quarkus, I suggest you browse their docs. So far, I have found them detailed, well written, and I will personally continue using them while I explore Quarkus further.

Creating a project

Quarkus provides a maven command to bootstrap your projects, for example (the below works for Linux and MacOS, see the Quarkus - Getting started guide for more information):

mvn io.quarkus:quarkus-maven-plugin:1.7.2.Final:create \
    -DprojectGroupId=dev.lankydan \
    -DprojectArtifactId=quarkus \
    -DclassName="dev.lankydan.web.people.PersonResource" \
    -Dpath="/hello"

This generates a basic project with a HTTP endpoint in PersonResource with the path hello, along with several resources to build and run your application. In fact, you should have enough at this point to run the generated code, both locally and inside a container.

A pom.xml is included that contains the required dependencies. You will need to add to this if you want to write anything past a hello world example.

By running (from inside your project):

./mvnw quarkus:list-extensions

You can see the Quarkus dependencies/extensions available to your project.

Then you can add an extension using a command like the one below:

./mvnw quarkus:add-extension -Dextensions="hibernate-validator"

The other option is to add it manually as I would typically do when sorting out my dependencies. Just thought I'd mention the add-extension command as I hadn't actually come across this method of dependency management when using Maven.

I keep mentioning Maven, but there is no reason you can't using something else, such as Gradle.

Quarkus uses JAX-RS for its REST capabilities

Quarkus follows the JAX-RS specification with Resteasy providing its implementation. If you have worked with Java for a reasonable time, you have probably seen or written code that follows the JAX-RS spec. And if you haven't, you probably will at some point. I haven't needed to use it before myself (I tend to use Spring when in Java land) so it is reasonably new to me. Anyway, in general, you can implement most of a simple REST API by relying on a few annotations.

Quarkus uses CDI beans for dependency injection

Quarkus' dependency solution is based on the CDI (Context and Dependency injection) specification. It should be noted that Quarkus does not provide a full CDI implementation, meaning that there will be limitations compared to other frameworks that you might have used. More information can be found in Quarkus' CDI reference and their Introduction to CDI.

Implementing a REST API

I have put together an example REST API that manages people. I mean, the purpose of the API doesn't really matter, it's just an example after all. Here comes a long chunk of code:

@Path("/people")
public class PersonResource {

  private final PersonRepository personRepository;

  public PersonResource(PersonRepository personRepository) {
    this.personRepository = personRepository;
  }

  @GET
  @Produces(MediaType.APPLICATION_JSON)
  public List<Person> all() {
    return personRepository.findAll();
  }

  @GET
  @Path("/{id}")
  @Produces(MediaType.APPLICATION_JSON)
  public Person get(@PathParam("id") UUID id) {
    return personRepository.findById(id);
  }

  @POST
  @Produces(MediaType.APPLICATION_JSON)
  public Person post(Person person) {
    return personRepository.insert(
        new Person(UUID.randomUUID(), person.getName(), person.getAge())
    );
  }

  @PUT
  @Path("/{id}")
  @Produces(MediaType.APPLICATION_JSON)
  public Person put(@PathParam("id") UUID id, Person person) {
    if (personRepository.findById(id) == null) {
      throw new PersonNotFoundException(id);
    }
    return personRepository.update(new Person(id, person.getName(), person.getAge()));
  }

  @DELETE
  @Path("/{id}")
  @Produces(MediaType.APPLICATION_JSON)
  public void delete(@PathParam("id") UUID id) {
    if (personRepository.findById(id) == null) {
      throw new PersonNotFoundException(id);
    }
    personRepository.deleteById(id);
  }
}

As I mentioned earlier, Quarkus uses Resteasy as its JAX-RS implementation. All of the annotations in this class are related to JAX-RS. Let's have a quick look at one of the functions:

@GET
@Path("/{id}")
@Produces(MediaType.APPLICATION_JSON)
public Person get(@PathParam("id") UUID id) {
  return personRepository.findById(id);
}

@GET specifies that the method handles GET HTTP requests.
@Path("/{id}") and @PathParam work together to take the HTTP path variable id and use it as the endpoint's input
@Produces(MediaType.APPLICATION_JSON) denotes the type of data that the endpoint returns as JSON

That's enough about annotations. The rest of the method consists of finding the Person matching the input id and returning it to whatever called the endpoint. Instead of returning a Person directly, you could use a Response object, but I wanted to keep it simple and easier to read.

JSON serialization

You have a choice between using JSON-B or Jackson for your serialization. Personally, I tend to use Jackson so I went ahead and plugged it in. You will need to add the extension for whichever you choose:

<dependency>
  <groupId>io.quarkus</groupId>
  <artifactId>quarkus-resteasy-jackson</artifactId>
</dependency>

Or via the command:

./mvnw quarkus:add-extension -Dextensions="resteasy-jackson"

Replace "jackson" with "jsonb" to switch over to JSON-B

Connecting to a database using JDBC

This API is not going to do anything useful if it isn't persisting data somewhere.

Quarkus provides JDBC implementations for connecting to databases. I chose to use Postgres for this project's database, but you can obviously use whatever suits your needs. More information can be found in Quarkus' data source documentation.

The first thing you need to add to connect to a database, is a JDBC driver dependency:

<dependency>
  <groupId>io.quarkus</groupId>
  <artifactId>quarkus-jdbc-postgresql</artifactId>
</dependency>

You could also run:

./mvnw quarkus:add-extension -Dextensions="jdbc-postgresql"

Then add properties specifying the connection type and the details (to your application.properties):

quarkus.datasource.db-kind=postgresql
# Change this part to match your connection details!
quarkus.datasource.jdbc.url=jdbc:postgresql://localhost:5432/mytestdb
quarkus.datasource.username=admin
quarkus.datasource.password=admin

Connection pooling

You don't want to directly manage the database connections yourself (I assume so anyway).

Quarkus' preferred JDBC data source and connection pooling library is Agroal. Below is a quote from the Quarkus data source documentation.

Agroal is a modern, light weight connection pool implementation designed for very high performance and scalability, and features first class integration with the other components in Quarkus, such as security, transaction management components, health metrics.

What this means for you:

Add the Agroal extension to your dependencies:

<dependency>
  <groupId>io.quarkus</groupId>
  <artifactId>quarkus-agroal</artifactId>
</dependency>

Or run:

./mvnw quarkus:add-extension -Dextensions="agroal"

Inject the DataSource that it initialises into your code (either directly via AgroalDataSource or using the DataSource interface).

Creating a repository

Once you have sorted out the data source connection and have enabled connection pooling, you can now start writing queries. Below is a slimmed-down version of the PersonRepository since its using raw JDBC meaning that the code is quite lengthy and duplicated between functions:

@ApplicationScoped
public class PersonRepository {

  private static final String FIND_BY_ID = "SELECT * FROM people WHERE id = ?";

  private final DataSource dataSource;

  public PersonRepository(DataSource dataSource) {
    this.dataSource = dataSource;
  }

  public Person findById(UUID id) {
    try (Connection connection = dataSource.getConnection();
        PreparedStatement statement = connection.prepareStatement(FIND_BY_ID)) {
      statement.setObject(1, id);
      try (ResultSet resultSet = statement.executeQuery()) {
        if (resultSet.next()) {
          return new Person(
              UUID.fromString(resultSet.getString("id")),
              resultSet.getString("name"),
              resultSet.getInt("age"));
        }
      }
    } catch (SQLException e) {
      throw new PersistenceException(e);
    }
    return null;
  }
}

The main take away of this snippet is the usage of the DataSource which Agroal provided the implementation for. You can inject it into your components after only needing to include the dependencies for a JDBC driver and Agroal itself. In regards to the JDBC code, more information on general JDBC usage can be found here.

Running the application

To run a Quarkus application, execute:

./mvnw compile quarkus:dev

This runs the application in dev mode, which enables hot-reloading. Making your development experience a bit smoother. Although, I did find it hard to break the habit of restarting my application after every code change...

To run your application for real, or non dev mode, you need to build the application's jar and run it. You can do this using:

./mvnw package
java -jar target/quarkus-1.0-SNAPSHOT-runner.jar

./mvnw package places the generated jar in the target directory

If you used the command I showed at the beginning of this post to generate your project, you would have a README.md containing all this information. Including how to construct a native executable and running your Quarkus application within a container. I will cover some of this in future blog posts.

Summary

You should now have a good idea on how to write a REST API using Quarkus. Like most other frameworks, you can leverage dependencies to quickly and effectively build your applications. Quarkus allows you to do the same. That can be seen by the number of times I told you to add a new dependency!

Bringing the focus back onto what Quarkus provides you, remember it's a framework specialising on fast start-up times and low memory usage. I purposely left further discussion on these factors out to allow the post to focus on building an application using Quarkus, which is pretty much the same as building any REST application that uses JAX-RS. So when developers/engineers, like you, search for how to build a REST API using Quarkus, they can see that it is nothing alien. I will cover the benefits of using Quarkus in future content

I have left parts out to not distract from the main content, but you can find all of the code I used to create the example in my GitHub repository.

If you enjoyed this post or found it helpful (or both) then please feel free to follow me on Twitter at @LankyDanDev and remember to share with anyone else who might find this useful!

Cancelling child coroutines

Dan Newton — Sun, 26 Jul 2020 00:00:00 +0000

Following on from my previous post, Cancelling coroutines, we will look at how cancelling a coroutine affects its children.

First let's clarify what a child is in this context:

runBlocking {
  val parent = launch(context = Dispatchers.Default) {
    val child = launch(context = Dispatchers.Default) {
      repeat(100) {
        println("[child] $it.. ")
        delay(50)
      }
    }
  }
}

The child is the second coroutine created in the snippet above. The coroutine whose scope the child was created in is the parent.

Cancelling a job cancels its children

Cancelling a parent coroutine/job will also cancel all of its child coroutines. Allowing you to submit a single cancellation request to stop many executing coroutines.

Below is example:

runBlocking {
  val parent = launch(context = Dispatchers.Default) {
    // children
    (1..2).map {
      launch(context = Dispatchers.Default) {
        repeat(100) { i ->
          println("[child $it] $i.. ")
          delay(50)
        }
      }
    }
    repeat(100) { i ->
      println("[parent] $i.. ")
      delay(50)
    }
  }
  delay(200)
  println("Cancelling")
  parent.cancel()
  parent.join()
  println("Finished cancelling job")
}

Which outputs:

[child 1] 0.. 
[parent] 0.. 
[child 2] 0.. 
[child 1] 1.. 
[child 2] 1.. 
[parent] 1.. 
[parent] 2.. 
[child 2] 2.. 
[child 1] 2.. 
[parent] 3.. 
[child 1] 3.. 
[child 2] 3.. 
Cancelling
Finished cancelling job

Both the parent and child coroutines here are cancelled by the single Job.cancel executed on the parent. You don't need to manually retrieve the children from the parent and cancel them yourself. The Kotlin coroutine library developers are kind enough to add this functionality for you. To be honest, you'd probably be a bit annoyed if this ability was not available. Oh, the highs and lows of library development!

Cancelling via a coroutine's scope

In the previous section the children were cancelled by a call to Job.cancel. You can achieve the same behaviour by calling CoroutineScope.cancel. Let's look at a quick example:

runBlocking {
  val parent = launch(context = Dispatchers.Default) {
    // children
    (1..2).map {
      launch(context = Dispatchers.Default) {
        repeat(500) { i ->
          println("[child $it] $i.. ")
          delay(50)
        }
      }
    }
    delay(200)
    println("Cancelling in scope")
    // Added [this@launch] to make things a bit clearer
    // You could call [cancel] by itself in this situation
    this@launch.cancel()
  }
  parent.join()
  println("Finished cancelling job")
}

CourtineScope.cancel is an alternate path to Job.cancel:

public fun CoroutineScope.cancel(cause: CancellationException? = null) {
  val job = coroutineContext[Job] ?: error("Scope cannot be cancelled because it does not have a job: $this")
  job.cancel(cause)
}

Depending on what you are doing, this might be advantageous, but my current opinion is that Job.cancel is likely to be more useful. For example, you can maintain a list of executing coroutines/jobs that you can cancel when needed. Whereas CoroutineScope.cancel needs to be triggered from inside the scope itself. Therefore restraining its usefulness. Note, this comes from someone who has not had tons of experience using coroutines out in the wild.

A benefit it does provide is a clear point where new coroutines will not start. After calling CoroutineScope.cancel, no children will be spawned. This allows you to make a decision, from within the scope, whether to continue of not. Achieving such clear cut behaviour like this from outside the scope would be difficult.

Summary

Coroutines can create parent-child relationships allowing a parent to cancel the child coroutines/jobs running within its context. This ability will enable you to easier control the execution of your coroutines. You can choose to cancel only a parent coroutine and allow the library to handle the rest. You can use either Job.cancel or CoroutineScope.cancel, even though under the hood they do the exact same thing.

If you enjoyed this post or found it helpful (or both) then please feel free to follow me on Twitter at @LankyDanDev and remember to share with anyone else who might find this useful!

Missing @EntityScan annotation in Spring Data Neo4j

Dan Newton — Mon, 20 Jul 2020 00:00:00 +0000

Not including the @EntityScan annotation and specifying the correct packages in your Spring Data Neo4j application can cause the following error:

java.lang.ArrayIndexOutOfBoundsException: Index 0 out of bounds for length 0
    at org.neo4j.ogm.context.EntityRowModelMapper.extractColumnValue(EntityRowModelMapper.java:75) ~[neo4j-ogm-core-3.2.12.jar:3.2.12]
    at org.neo4j.ogm.context.EntityRowModelMapper.map(EntityRowModelMapper.java:64) ~[neo4j-ogm-core-3.2.12.jar:3.2.12]
    at org.neo4j.ogm.session.delegates.ExecuteQueriesDelegate.lambda$executeAndMap$1(ExecuteQueriesDelegate.java:151) ~[neo4j-ogm-core-3.2.12.jar:3.2.12]
    at org.neo4j.ogm.session.Neo4jSession.doInTransaction(Neo4jSession.java:590) ~[neo4j-ogm-core-3.2.12.jar:3.2.12]
    at org.neo4j.ogm.session.Neo4jSession.doInTransaction(Neo4jSession.java:564) ~[neo4j-ogm-core-3.2.12.jar:3.2.12]
    at org.neo4j.ogm.session.delegates.ExecuteQueriesDelegate.executeAndMap(ExecuteQueriesDelegate.java:138) ~[neo4j-ogm-core-3.2.12.jar:3.2.12]
    at org.neo4j.ogm.session.delegates.ExecuteQueriesDelegate.query(ExecuteQueriesDelegate.java:111) ~[neo4j-ogm-core-3.2.12.jar:3.2.12]
    at org.neo4j.ogm.session.Neo4jSession.query(Neo4jSession.java:425) ~[neo4j-ogm-core-3.2.12.jar:3.2.12]
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:na]
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:na]
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:na]
    at java.base/java.lang.reflect.Method.invoke(Method.java:566) ~[na:na]
    at org.springframework.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:282) ~[spring-core-5.2.7.RELEASE.jar:5.2.7.RELEASE]
    at org.springframework.data.neo4j.transaction.SharedSessionCreator$SharedSessionInvocationHandler.lambda$invoke$1(SharedSessionCreator.java:121) ~[spring-data-neo4j-5.3.1.RELEASE.jar:5.3.1.RELEASE]
    at org.springframework.data.neo4j.transaction.SharedSessionCreator$SharedSessionInvocationHandler.invokeInTransaction(SharedSessionCreator.java:159) ~[spring-data-neo4j-5.3.1.RELEASE.jar:5.3.1.RELEASE]
    at org.springframework.data.neo4j.transaction.SharedSessionCreator$SharedSessionInvocationHandler.invoke(SharedSessionCreator.java:123) ~[spring-data-neo4j-5.3.1.RELEASE.jar:5.3.1.RELEASE]
    at com.sun.proxy.$Proxy62.query(Unknown Source) ~[na:na]
    at org.springframework.data.neo4j.repository.query.GraphQueryExecution$CollectionExecution.execute(GraphQueryExecution.java:97) ~[spring-data-neo4j-5.3.1.RELEASE.jar:5.3.1.RELEASE]
    at org.springframework.data.neo4j.repository.query.GraphRepositoryQuery.doExecute(GraphRepositoryQuery.java:76) ~[spring-data-neo4j-5.3.1.RELEASE.jar:5.3.1.RELEASE]
    at org.springframework.data.neo4j.repository.query.AbstractGraphRepositoryQuery.execute(AbstractGraphRepositoryQuery.java:57) ~[spring-data-neo4j-5.3.1.RELEASE.jar:5.3.1.RELEASE]
    at org.springframework.data.repository.core.support.QueryExecutorMethodInterceptor$QueryMethodInvoker.invoke(QueryExecutorMethodInterceptor.java:195) ~[spring-data-commons-2.3.1.RELEASE.jar:2.3.1.RELEASE]
    at org.springframework.data.repository.core.support.QueryExecutorMethodInterceptor.doInvoke(QueryExecutorMethodInterceptor.java:152) ~[spring-data-commons-2.3.1.RELEASE.jar:2.3.1.RELEASE]
    at org.springframework.data.repository.core.support.QueryExecutorMethodInterceptor.invoke(QueryExecutorMethodInterceptor.java:130) ~[spring-data-commons-2.3.1.RELEASE.jar:2.3.1.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) ~[spring-aop-5.2.7.RELEASE.jar:5.2.7.RELEASE]
    at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:367) ~[spring-tx-5.2.7.RELEASE.jar:5.2.7.RELEASE]
    at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:118) ~[spring-tx-5.2.7.RELEASE.jar:5.2.7.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) ~[spring-aop-5.2.7.RELEASE.jar:5.2.7.RELEASE]
    at org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:139) ~[spring-tx-5.2.7.RELEASE.jar:5.2.7.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) ~[spring-aop-5.2.7.RELEASE.jar:5.2.7.RELEASE]
    at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:95) ~[spring-aop-5.2.7.RELEASE.jar:5.2.7.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) ~[spring-aop-5.2.7.RELEASE.jar:5.2.7.RELEASE]
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:212) ~[spring-aop-5.2.7.RELEASE.jar:5.2.7.RELEASE]
    at com.sun.proxy.$Proxy76.findAll(Unknown Source) ~[na:na]

Yes, there is a lot there, and you will need to scroll across to see it. The primary information to extract is that there is an ArrayIndexOutOfBoundsException when trying to map a query's result to an object, as denoted by EntityRowModelMapper.map and EntityRowModelMapper.extractColumnValue.

The error doesn't really tell you much.

Thankfully, there is a high probability that you are missing the @EntityScan annotation. Adding this to one of your configuration files or your main class (if you're using Spring Boot, its likely to be the class you annotated with @SpringBootApplication) and specifying the basePackages or basePackageClasses should tell Spring where to find your entity and relationship classes.

You are more likely to see the error above if you are using Spring Boot and do not have your @SpringBootApplication class in one of the root packages of your application's code. If this is the case, and you are using Spring Data's inferred queries, you will also need to include the @EnableNeo4jRepositories annotation. It will face the same style of problem as the missing @EntityScan.

For clarity below is a quick example:

package dev.lankydan.neo4j.application;

// Already needed to specify packages for other parts of the application
@SpringBootApplication(scanBasePackages = "dev.lankydan.neo4j")
@EnableNeo4jRepositories(basePackages = "dev.lankydan.neo4j")
public class Application implements CommandLineRunner {

  public static void main(String[] args) {
    SpringApplication.run(Application.class);
  }
}

This is the main class of the application, note that the Application class is located in dev.lankydan.neo4j.application.

Now for the entity itself:

package dev.lankydan.neo4j.entity;

@NodeEntity(value = "City")
public class City {
  // contents of the class
}

The entity has been placed in dev.lankydan.neo4j.entity, which is not a sub-package of dev.lankydan.neo4j.application. Therefore, Application needs to be adjusted, giving the final iteration:

package dev.lankydan.neo4j.application;

// Already needed to specify packages for other parts of the application
@SpringBootApplication(scanBasePackages = "dev.lankydan.neo4j")
@EnableNeo4jRepositories(basePackages = "dev.lankydan.neo4j")
// The [basePackages] could be specific like the one below or more general like the declarations above
@EntityScan(basePackages = "dev.lankydan.neo4j.entity")
public class Application implements CommandLineRunner {

  public static void main(String[] args) {
    SpringApplication.run(Application.class);
  }
}

Spring can now detect the entity classes and allow the application to run smoothly. A second option would be to move Application to a different package, closer to the root of the project. In this scenario, moving it to dev.lankydan.neo4j would do the trick.

Hopefully, this will help you remedy your code's issue.

If you enjoyed this post or found it helpful (or both) then please feel free to follow me on Twitter at @LankyDanDev and remember to share with anyone else who might find this useful!

Mapping a path query in Spring Data Neo4j

Dan Newton — Sun, 19 Jul 2020 00:00:00 +0000

Spring data does a lot to help you focus on writing your cypher queries while it handles mapping the results for you. However, when your queries become more complex, it starts to struggle. This is where you need to step in and explicitly map the query results into your domain objects.

In this post, we will look at how you can query a path and map the results using Spring Data Neo4j.

Simple queries are mapped for you

For simple queries, you can get away with only defining the relevant entity and relationship classes, Spring will do the rest.

Let's use a findAll function (returns all nodes):

@Query("MATCH (n:City) RETURN n LIMIT 25")
Iterable<City> findAll();

The City entity needs to defined, as such:

@NodeEntity(value = "City")
public class City {

  @Id
  @GeneratedValue
  private Long id;
  private String name;
  private double longitude;
  private double latitude;

  @Relationship(type = "NEXT")
  private Set<Connection> connections = Collections.emptySet();

  public City(
      Long id,
      String name,
      double longitude,
      double latitude,
      Set<Connection> connections
  ) {
    this.id = id;
    this.name = name;
    this.longitude = longitude;
    this.latitude = latitude;
    this.connections = connections;
  }

  public City() { }

  // getters + setters
}

City also has a relationship, so that needs to be defined as well:

@RelationshipEntity("NEXT")
public class Connection {

  @Id
  @GeneratedValue
  private long id;
  @StartNode
  private City start;
  @EndNode
  private City end;

  public Connection(long id, City start, City end) {
    this.id = id;
    this.start = start;
    this.end = end;
  }

  public Connection() { }

  // getters + setters
}

Calling the findAll function will correctly collect the cities from the database and map them for you.

Everything is good at this point.

Another example of setting up a Neo4j entity class can be found in the Spring Data Neo4j - getting started documentation.

Extracting paths requires your assistance

The code above is not suitable for a cypher query that returns a path.

The following query will be used for this section (taken from the Apoc plugin's documentation):

MATCH (a:City {name:'bruges'}), (b:City {name:'dresden'})
MATCH p=(a)-[*]->(b)
WITH collect(p) as paths
CALL apoc.spatial.sortByDistance(paths) YIELD path, distance
RETURN path, distance

This query returns the path between two cities and the total distance of that path.

Unfortunately, there is no way to represent this information using the entity and relationship classes defined earlier. Furthermore, there is no way to currently do this using Spring Data Neo4j full stop. It personally took me a long time to realise this fact, until I came across this excerpt in their documentation.

Custom queries do not support a custom depth. Additionally, @Query does not support mapping a path to domain entities, as such, a path should not be returned from a Cypher query. Instead, return nodes and relationships to have them mapped to domain entities.

Somehow I managed to keep brushing over this information when I went through the docs...

This means you cannot do something like this:

public interface PathRepository extends Neo4jRepository<City, Long> {

  @Query("MATCH (a:City {name:$departure}), (b:City {name:$arrival})\n"
      + "MATCH p=(a)-[*]->(b)\n"
      + "WITH collect(p) as paths\n"
      + "CALL apoc.spatial.sortByDistance(paths) YIELD path, distance\n"
      + "RETURN path, distance")
  List<Path> getAllPaths(String departure, String arrival);
}

Where Path contains the following:

@QueryResult
public class Path {
  public List<City> path;
  public double distance;

  public Path(List<City> path, double distance) {
    this.path = path;
    this.distance = distance;
  }
}

The @QueryResult annotation allows you to define a custom object to contain the results of your cypher query, more can be found in the Spring docs

If you do this, it will still execute, but there will be no information about the cities filled in.

How to manually map a path

To be able to map the path shown in the previous section, you need to manually assign the results from the cypher query into objects.

This can be done with the following code (large snippet incoming):

@Repository
// Class instead of interface
// Extend [SimpleNeo4jRepository] instead of [Neo4jRepository]
public class PathRepository extends SimpleNeo4jRepository<City, Long> {

  private static final String GET_ALL_PATHS_QUERY =
      "MATCH (a:City {name:$departure}), (b:City {name:$arrival})\n"
          + "MATCH p=(a)-[*]->(b)\n"
          + "WITH collect(p) as paths\n"
          + "CALL apoc.spatial.sortByDistance(paths) YIELD path, distance\n"
          + "RETURN path, distance";

  // Needed to be able to query the database
  private final Session session;

  // Inject in the session
  // No need to create the session yourself, Spring has already created it
  public PathRepository(Session session) {
    super(City.class, session);
    this.session = session;
  }

  public List<Path> getAllPaths(String departure, String arrival) {
    Map<String, String> parameters = Map.of(
        "departure", departure,
        "arrival", arrival
    );

    // Execute the query and retrieve the result
    Result rows = session.query(GET_ALL_PATHS_QUERY, parameters);

    List<Path> results = new ArrayList<>();
    for (Map<String, Object> row : rows) {
      results.add(convertRow(row));
    }

    return results;
  }

  private Path convertRow(Map<String, Object> row) {
    InternalPath.SelfContainedSegment[] connections =
        (InternalPath.SelfContainedSegment[]) row.get("path");

    List<City> cities = new ArrayList<>();
    // Iterate through the segments in the path
    for (InternalPath.SelfContainedSegment connection : connections) {
      cities.add(convert(connection));
    }

    double distance = (Double) row.get("distance");
    return new Path(cities, distance);
  }

  private City convert(InternalPath.SelfContainedSegment connection) {
    // Extract the information about the [City] from the path segment
    // Information about the start node and the relationship could also be accessed
    return new City(
        connection.end().id(),
        connection.end().get("name").asString(),
        connection.end().get("latitude").asDouble(),
        connection.end().get("longitude").asDouble()
    );
  }
}

This iteration of PathRepository extends SimpleNeo4jRepository to inherit some of the more common queries without requiring you to implement them yourself. You do not need to extend this class, but I suggest you do.

I don't think there is any need to explain the rest of the example, I tidied up the code as best I could and added comments for clarity. Hopefully, they are enough!

Conclusion

As I eventually found out and have now mentioned to you, at the time of writing this post, Spring Data Neo4j does not support the automatic mapping of queries containing paths. Therefore you will need to take your fate into your own hands and extract the data yourself. The code in the penultimate section (the long code snippet) shows you how to do this. Using a similar structure, you should be able to adapt this for your own use to solve your own problems. I hope this saves you some time as it took me a while to realise this was the only way to retrieve a path, when using Spring Data Neoj4 anyway...

If you enjoyed this post or found it helpful (or both) then please feel free to follow me on Twitter at @LankyDanDev and remember to share with anyone else who might find this useful!