Article

How to use the new Kafka Client API for Kafka-specific message properties in Open Liberty 20.0.0.3

March 17, 2020
Event-DrivenJava
Tom Jennings
Software Developer and Adocate Intern
Open Liberty

In Open Liberty 20.0.0.3, you can now access Kafka-specific properties such as the message key and message headers, rather than just the message payload, as was the case with the basic MicroProfile Reactive Messaging Message API. Also, you can now set the SameSite attribute in the session cookie, the LTPA, and JWT cookies as well as in application-defined cookies.

Note: View the list of fixed bugs in Open Liberty 20.0.0.3 here.

If you're interested in what's coming soon in Open Liberty, take a look at our current development builds. These include updated features to MicroProfile (Rest Client, Metrics, Health, Fault Tolerance, and Config), as well as GraphQL with Open Liberty, automatically compressing HTTP responses, and persistent EJB timers.

Run your apps using 20.0.0.3

If you're using Maven, here are the coordinates:

<dependency>
    <groupId>io.openliberty</groupId>
    <artifactId>openliberty-runtime</artifactId>
    <version>20.0.0.3</version>
    <type>zip</type>
</dependency>

Or, for Gradle:

dependencies {
    libertyRuntime group: 'io.openliberty', name: 'openliberty-runtime', version: '[20.0.0.3,)'
}

Or, if you're using Docker:

FROM open-liberty

Or, take a look at our downloads page.

Ask a question on Stack Overflow

Kafka-specific properties

New to Open Liberty is Kafka-specific properties. The basic MicroProfile Reactive Messaging Message API does not let the user access anything other than the message payload. The native Kafka client API allows users to access Kafka-specific message properties, such as the message key and message headers.

Incoming messages

For incoming messages, we have now allowed the user to unwrap a Message to gain access to the underlying ConsumerRecord:

>@Incoming("channel1")
public CompletionStage consume(Message message) {
    ConsumerRecord<String, String> cr = (ConsumerRecord<String, String>) message.unwrap(ConsumerRecord.class);
    String key = consumerRecord.key();
    String value = consumerRecord.value();
    String topic = consumerRecord.topic();
    int partition = consumerRecord.partition();
    long timestamp = consumerRecord.timestamp();
    Headers headers = consumerRecord.headers();
    // some more code....
    return CompletableFuture.completedFuture(null);
}

Outgoing messages

For outgoing messages, if the payload is a ProducerRecord, then the properties within it are passed on to Kafka:

>@Outgoing("channel2")
public Message publish() throws UnsupportedEncodingException {
   ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("myTopic", null, "myKey", "myValue");
   producerRecord.headers().add("HeaderKey", "HeaderValue".getBytes("UTF-8"));
   return Message.of(producerRecord);
}

The example above assumes that no topic was explicitly pre-configured in the MicroProfile Config for the channel. If the topic is pre-configured, then that will take precedence and the topic in the ProducerRecord will be ignored.

Example using a pre-configured topic

In this example, the topic is pre-configured using MicroProfile Config to be myOtherTopic, so the topic set in the ProducerRecord is ignored. Here are the MicroProfile Config properties:

>mp.reactive.messaging.channel3.connector=liberty-kafka
mp.reactive.messaging.channel3.topic=myOtherTopic #Overrides value in code

Here is the reactive messaging bean:

>@Outgoing("channel3")
public Message<ProducerRecord<K, V>> publish() {
   ProducerRecord pr = new ProducerRecord("myTopic", "myValue");
   return Message.of(pr);
}/pre>

Adding the SameSite cookie attribute

Open Liberty now offers the ability to set the SameSite attribute on the session cookie, the LTPA, and the JWT cookies as well as on the application-defined cookies. Implement the SameSite attribute by adding one or more server.xml configuration options. The servlet javax.servlet.http.Cookie API does not offer the ability to set the SameSite attribute on a cookie.

If the SameSite attribute is needed, the options for setting it are currently limited to using the HttpServletResponse.addHeader and HttpServletResponse.setHeader, and constructing the Set-Cookie header. The SameSite attribute is used by browsers to determine if a particular cookie should be sent with a request. There are three values for the SameSite attribute: Lax, Strict, and None, as shown in Table 1.

Table 1: Values for the SameSite attribute.
Value Description
SameSite=Strict The cookie will only be sent along with same-site requests. The cookie is only sent by the browser if the site for the cookie matches the site in the address bar, for example.
SameSite=Lax The cookie will be sent with same-site requests and with cross-site top-level navigations. Clicking a link, for example.
SameSite=None The cookie will be sent with same-site and cross-site requests. The cookie is sent by the browser for third party contexts and embedded content, for example.

To use the SameSite cookie attribute:

    1. Set the session cookie SameSite attribute using the following server.xml configuration:

<httpSession cookieSameSite="Disabled|Strict|Lax|None"/>

The default value is Disabled. This means that no SameSite attribute will be added.

    1. Set the LTPA/JWT cookie SameSite attribute using the following server.xml configuration:

<webAppSecurity sameSiteCookie="Disabled|Strict|Lax|None"/>

The default value is Disabled. This means that no SameSite attribute will be added.

    1. Set the SameSite attribute on cookies via the following server.xml configuration:
<httpEndpoint id="defaultHttpEndpoint" 
                httpPort="9080" 
                httpsPort="9443" >
   <samesite lax="cookieOne" strict="cookieTwo" none="cookieThree"/>
</httpEndpoint>

The <httpEndpoint/> SameSite configuration allows the use of wildcards in the following ways:

  • A standalone wildcard ( * ). All cookies would have the SameSite=Lax attribute, which includes the session and LTPA/JWT cookies unless the <httpSession/> and/or <webAppSecurity/> configuration has also been set. For example:
     <httpEndpoint id="defaultHttpEndpoint" httpPort="9080" httpsPort="9443" >
        <samesite lax="*" />
     </httpEndpoint>
  • At the end of one or more cookie names. The following snippet would map the following cookie names to SameSite attributes:
    • cookieOne → SameSite=Lax
    • cookieTwo → SameSite=Strict
    • cookieThree → SameSite=None
  <httpEndpoint id="defaultHttpEndpoint" httpPort="9080" httpsPort="9443" >
        <samesite lax="cookie*" strict="cookieTwo" none="cookieThree"/>
  </httpEndpoint>

The <httpSession/> and <webAppSecurity/> configuration takes precedence over the <httpEndpoint/>configuration>.

  • When a cookie matches the SameSite=None configuration, then the Secure attribute will automatically be added to the cookie.

Note: The <httpEndpoint/> configuration can apply to any Set-Cookie header.

Technical details regarding the SameSite attribute can be found in the following RFC: Cookies: HTTP State Management Mechanism.

Try Open Liberty 20.0.0.3 in Red Hat Runtimes now

Open Liberty is part of the Red Hat Runtimes offering. If you're a Red Hat Runtimes subscriber, you can try Open Liberty now.

Last updated: February 13, 2024