DEV Community: yangbongsoo

Self Review(2023) & New Challenge

yangbongsoo — Sun, 19 May 2024 05:12:18 +0000

Self Review

2023 was the year we completed the transition project for our online commerce market order/payment system. It took a total of three years. As a result, all the feedback from my colleagues was positive. Reflecting on the hard work, I also found myself saying only good things.

The top keywords in my peer reviews were passion, collaboration/contribution, and sharing/feedback. Here are some excerpts from my colleagues' feedback. Although I feel a bit embarrassed by the high praise, it feels good to be recognized for my hard work.

colleagues feedback

Even when tasks might be tedious, you spare no effort in ensuring that your colleagues fully understand and acquire the necessary knowledge during work sharing. Thanks to this, we were able to disseminate essential project maintenance knowledge more quickly across various relevant departments.
You structurally understand the parts of the project that need discussion and actively share this information with the team.
Your proactive communication style helps tasks that could potentially be blocked to proceed faster than expected.
Your exemplary attitude and broad activity are a great asset to the team. Leading from the front inspires team members significantly, and your ability to find solutions when problems arise is outstanding. Your fearless approach to challenges contributes greatly to the team's achievements. Your diverse roles, based on extensive activity, enhance our team's versatility.
It is impressive how you voluntarily choose and complete tasks that others find difficult to tackle.

Areas for Improvement

In 2023, I worked harder and more proactively than anyone else, but once the transition was complete, I quickly felt exhausted, both physically and mentally, like someone crossing the finish line. So, I used all my annual leave in December and rested for three weeks, but even after returning, I didn't fully recover.

In January of this year, an issue with the order system occurred due to the sale of merchandise by a major streamer. I felt responsible and did my best with the follow-up actions. However, overall, my sense of responsibility was somewhat lacking. Although I resolved to do my best until the end, I didn't actually follow through.

Last year, if I encountered an issue I couldn't resolve, I would contact stakeholders directly to try to solve it, and if that didn't work, I would quickly inform my leader to ask for help. This year, however, I waited until the other party contacted me.

Looking back now, by the end of last year, my energy was already depleted, and I was just holding on with the sole focus of completing the new order system transition. During work hours, I often kept the monitoring system running, refreshing the screen every five seconds, anxiously standing guard. That process was the hardest part. Watching the screen flicker continuously made me feel like I was going to throw up. After spending a couple of months like that, my physical and mental strength deteriorated, and my sense of responsibility crumbled easily once the transition was over.

However, after completing the follow-up actions for the January order system issue, I had another chance to be evaluated. The streamer decided to do an additional sale in April, and that monitoring task was my last. Fortunately, this time the sale proceeded smoothly and stably.

New Challenge

I am moving to the Storage Platform team to develop an object storage service (similar to AWS S3). Previously, in this task force, I collaborated with Naver Financial Corporation, and moving forward, I will be working with Naver Cloud Corporation.

The Storage Platform team develops in golang and is closely involved with low-level development like file systems. While many things are unfamiliar, including the programming language, I feel confident that I can adapt well. My successful experience in previous projects gives me this confidence.

Many people have asked why I am moving to Storage Platform team. Since 2022, I have been attending graduate school. I have taken courses in distributed systems, Linux OS, file systems, and databases, and I enjoyed studying computer science through these classes.

Unlike my undergraduate years, having more development experience made it easier and more enjoyable to understand the course material. I asked my professors many questions, so much so that a friend in my classes jokingly called me Hermione. I paid $4,500 per semester, so it felt like a waste to just attend classes without fully engaging. Additionally, I knew it would be difficult to have such opportunities to ask professors questions later on.

Previously, the Storage Platform team felt like a wall, but taking these graduate courses gave me the courage to make the move. While I may not immediately handle core tasks, I dream of a day when, with consistent effort, I can contribute to file system development.

Grouping the Same Products in Java

yangbongsoo — Sun, 14 Apr 2024 03:23:43 +0000

First, one must understand the relationship between Product and Item. A Product can have multiple Items. Simply put, a Product is the product information, and an Item is the option information.

import java.util.List;

import lombok.Builder;
import lombok.Value;

@Value
@Builder
public class Product {
    String productId;

    List<Item> items;

    @Value
    @Builder
    public static class Item {
        String itemId;

        long itemQuantity;
    }
}

In the sample products, there are a total of three Products, two of which have the same productId. At the Item level, itemId 1 and 2 also exist as duplicates. Ultimately, the goal is to group by the same productId and the same itemId.

public class Main {
  public static void main(String[] args) {
    /**
     * [
     *   Product(
     *     productId=1,
     *     items=[
     *       Product.Item(itemId=1, itemQuantity=1),
     *       Product.Item(itemId=1, itemQuantity=1),
     *       Product.Item(itemId=2, itemQuantity=2)
     *     ]
     *   ),
     *   Product(
     *     productId=1,
     *     items=[
     *       Product.Item(itemId=3, itemQuantity=3),
     *       Product.Item(itemId=2, itemQuantity=2)
     *     ]
     *   ),
     *   Product(
     *     productId=2,
     *     items=[
     *       Product.Item(itemId=5, itemQuantity=5)
     *     ]
     *   )
     * ]
     */
     List<Product> products = getProducts();
  }

  private static List<Product> getProducts() {
    return List.of(
      Product.builder()
        .productId("1")
        .items(List.of(
          Item.builder()
            .itemId("1")
            .itemQuantity(1)
            .build(),
          Item.builder()
            .itemId("1")
            .itemQuantity(1)
            .build(),
          Item.builder()
            .itemId("2")
            .itemQuantity(2)
            .build()
        )
      )
      .build(),
      Product.builder()
        .productId("1")
        .items(List.of(
          Item.builder()
            .itemId("3")
            .itemQuantity(3)
            .build(),
          Item.builder()
            .itemId("2")
            .itemQuantity(2)
            .build()
        )
      )
      .build(),
      Product.builder()
        .productId("2")
        .items(List.of(
          Item.builder()
            .itemId("5")
            .itemQuantity(5)
            .build()
          )
        )
      .build()
    );
  }
}

The ProductGroup class below holds the results of grouping by the same productId and the same itemId. It can have multiple ItemGroups per productId without duplicates, and the itemId within the ItemGroup also does not duplicate. Finally, the itemQuantity for the same itemId is summed up.

@Value
@Builder(access = AccessLevel.PRIVATE)
public class ProductGroup {
    String productId;

    List<ItemGroup> itemGroups;

    @Value
    @Builder(access = AccessLevel.PRIVATE)
    public static class ItemGroup {
        String itemId;

        long totalQuantity;
    }
}

Let's now look at the logic for grouping. The first thing to do is to group the Products by productId.

public static List<ProductGroup> grouping(List<Product> products) {
  return products.stream()
    // first process Map<String, List<Product>>
    .collect(Collectors.groupingBy(Product::getProductId, toList()))


    .entrySet() // Set<Map<K,V>.Entry<String, List<Product>>>
    .stream() // Stream<Map<K,V>.Entry<String, List<Product>>>

    // second process
    .map(ProductGroup::makeEachProductGroups) // Stream<ProductGroup>
    .toList();
}

After performing the first process, it is stored in a Map as described in the comment below. Using this Map, a ProductGroup is created by utilizing entrySet().stream().

public static List<ProductGroup> grouping(List<Product> products) {
  return products.stream()

    /**
     * first process : Map<String, List<Product>> productListsById
     * {
     * 1=[
     *     Product(
     *       productId=1,
     *       items=[
     *         Product.Item(itemId=1, itemQuantity=1),
     *         Product.Item(itemId=1, itemQuantity=1),
     *         Product.Item(itemId=2, itemQuantity=2)
     *       ]
     *     ),
     *     Product(
     *       productId=1,
     *       items=[
     *         Product.Item(itemId=3, itemQuantity=3),
     *         Product.Item(itemId=2, itemQuantity=2)
     *       ]
     *     )
     *   ],
     * 2=[
     *     Product(
     *       productId=2,
     *       items=[Product.Item(itemId=5, itemQuantity=5)]
     *     )
     *   ]
     * }
     */
    .collect(Collectors.groupingBy(Product::getProductId, toList()))

    .entrySet() // Set<Map<K,V>.Entry<String, List<Product>>>
    .stream() // Stream<Map<K,V>.Entry<String, List<Product>>>

    // second process
    .map(ProductGroup::makeEachProductGroups) // Stream<ProductGroup>
    .toList();

cf) If toUnmodifiableMap is used as shown below, a duplicateKeyException occurs if there are duplicate productIds.

Map<String, Product> productsById = products.stream()
.collect(Collectors.toUnmodifiableMap(Product::getProductId, Function.identity()));

In the subsequent secondary task, a ProductGroup is created. Then, moving on to the ItemGroup stage (makeItemGroups), the Items within a Product are first turned into a Stream using flatMap(product -> product.getItems().stream()), and then grouped again by itemId.

Using the results, an ItemGroup is created, and the totalQuantity is calculated by summing the quantities of the grouped Items.

private static ProductGroup makeEachProductGroups(Entry<String, List<Product>> productListById) {
  return ProductGroup.builder()
    .productId(productListById.getKey())
    .itemGroups(makeItemGroups(productListById.getValue()))
    .build();
  }

private static List<ItemGroup> makeItemGroups(List<Product> products) {
  return products.stream() // Stream<Product>
    .flatMap(product -> product.getItems().stream()) // Stream<Product.Item>
    .collect(groupingBy(Item::getItemId, toList())) // Map<String, List<Product.Item>>
    .entrySet().stream() // Stream<Map<K,V>.Entry<String, List<Product.Item>>>
    .map(it ->
      ItemGroup.builder()
        .itemId(it.getKey())
        .totalQuantity(it.getValue().stream().mapToLong(Item::getItemQuantity).sum())
        .build()
    ) // Stream<ProductGroup.ItemGroup>
    .toList();
  }

The final result obtained from the grouping is as follows.

    /**
     * second process : grouping
     * [
     *   ProductGroup(
     *     productId=1,
     *     itemGroups=[
     *       ProductGroup.ItemGroup(itemId=1, totalQuantity=2),
     *       ProductGroup.ItemGroup(itemId=2, totalQuantity=4),
     *       ProductGroup.ItemGroup(itemId=3, totalQuantity=3)
     *     ]
     *   ),
     *   ProductGroup(
     *     productId=2,
     *     itemGroups=[
     *       ProductGroup.ItemGroup(itemId=5, totalQuantity=5)
     *     ]
     *   )
     * ]
     */

There's one important point to note in the makeItemGroups logic. After turning the Items within a Product into a Stream using flatMap(product -> product.getItems().stream()), it is essential to group them again by itemId.

It should not be done by grouping using the items of each product as follows.

private static List<ItemGroup> makeItemGroupsWrongWay(List<Product> products) {
  return products.stream()
    .flatMap(product -> // don't do that each product
      product.getItems().stream()
        .collect(groupingBy(Item::getItemId, toList()))
        .entrySet()
        .stream()
        .map(it ->
          ItemGroup.builder()
            .itemId(it.getKey())
            .totalQuantity(it.getValue().stream().mapToLong(Item::getItemQuantity).sum())
            .build()
        ))
        .toList();
}

Test based on Enum type(DynamicTest)

yangbongsoo — Sun, 07 Apr 2024 05:47:41 +0000

When developing e-commerce, one encounters numerous Enums. The following Enum represents a simple product type code. There are two types: simple options and complex options.

public enum ProductType {
    SIMPLE_OPTION,
    COMPLEX_OPTION
}

To test the parts that vary according to the product type, we can use @ParameterizedTest and @EnumSource.

@ParameterizedTest
@EnumSource(names = {"SIMPLE_OPTION", "COMPLEX_OPTION"})
void enumSourceTest(ProductType productType) {
    System.out.println(productType);
}

However, it's very common for business logic to branch based on the relationships between Enums, and one may want to test those parts.

The code below represents the relationship between the product type at the product level and the item type at the item level. While there are eight possible combinations of ProductType and ItemType, only four of these combinations are actually feasible.

@Value
@Builder
public class Product {
    ProductType productType;

    List<Item> items;

    @Value
    @Builder
    public static class Item {
        ItemType itemType;
    }
}

// The relationship between ProductType and ItemType
// SIMPLE_OPTION - SIMPLE
// COMPLEX_OPTION - COMBINATION1
// COMPLEX_OPTION - COMBINATION2
// COMPLEX_OPTION - COMBINATION3
public enum ProductType {
    SIMPLE_OPTION,
    COMPLEX_OPTION
}

public enum ItemType {
    SIMPLE,
    COMBINATION1,
    COMBINATION2,
    COMBINATION3,
}

Therefore, we need to set up these four combinations separately, which can be done using @ParameterizedTest and @MethodSource.

@ParameterizedTest
@MethodSource("productTypeAndItemTypeSet")
void methodSourceTest(ProductType productType, ItemType itemType) {
    System.out.println(productType + " : " + itemType);
}

private static Stream productTypeAndItemTypeSet() {
    return Stream.of(
        Arguments.of(
            ProductType.SIMPLE_OPTION,
            ItemType.SIMPLE
        ),
        Arguments.of(
            ProductType.COMPLEX_OPTION,
            ItemType.COMBINATION1
        ),
        Arguments.of(
            ProductType.COMPLEX_OPTION,
            ItemType.COMBINATION2
        ),
        Arguments.of(
            ProductType.COMPLEX_OPTION,
            ItemType.COMBINATION3
        )
    );
}

There's also a way to utilize org.junit.jupiter.api.DynamicTest and @TestFactory.

Let's look at the code first. It sets up the four Enum combinations with Stream.of and then runs each through DynamicTest.dynamicTest within a flatMap structure.

@TestFactory // It's not @Test but @TestFactory
Stream<DynamicTest> dynamicTestTest() {
    return Stream.of(
            Map.of(ProductType.SIMPLE_OPTION, ItemType.SIMPLE),
            Map.of(ProductType.COMPLEX_OPTION, ItemType.COMBINATION1),
            Map.of(ProductType.COMPLEX_OPTION, ItemType.COMBINATION2),
            Map.of(ProductType.COMPLEX_OPTION, ItemType.COMBINATION3)
        )
        .flatMap(testCaseMap ->
            testCaseMap.entrySet().stream()
                .map(it ->
                    DynamicTest.dynamicTest(testCaseMap.toString(), () -> {
                        ProductType productType = it.getKey();
                        ItemType itemType = it.getValue();

                        System.out.println(productType + " : " + itemType);
                        }
                    )
                )
        );
    }
}

Java doesn't provide a Pair class by default, so using Map made the implementation a bit complicated, requiring the use of flatMap.

Creating it with Kotlin and using Pair would indeed make it simpler.

@TestFactory
fun dynamicTestTest(): Stream<DynamicTest> {
    return Stream.of(
        Pair(ProductType.SIMPLE_OPTION, ItemType.SIMPLE),
        Pair(ProductType.COMPLEX_OPTION, ItemType.COMBINATION1),
        Pair(ProductType.COMPLEX_OPTION, ItemType.COMBINATION2),
        Pair(ProductType.COMPLEX_OPTION, ItemType.COMBINATION3),
    )
        .map { pair ->
            DynamicTest.dynamicTest(pair.toString()) {
                println("${pair.first} : ${pair.second}")
            }
        }
}

cf) While Java doesn't provide a Pair class by default, we can use org.springframework.data.util.Pair provided by Spring.

WebClient Encoding Duplication Issue

yangbongsoo — Sun, 31 Mar 2024 03:06:17 +0000

When using WebClient's uriBuilder, it results in a doubly encoded output because it encodes an already encoded query string value again. In other words, there's no need to encode in advance.

// If constructed as below, the final request is sent as a doubly encoded result.
this.webClient
  .get()
    .uri(uriBuilder -> uriBuilder
      .path("/api")
      //.query("name=양봉수") It's korea language
      .query("name=%EC%96%91%EB%B4%89%EC%88%98")
      .build()
     )
  .retrieve()

However, if the query string value is of JSON type, it must be encoded in advance; otherwise, an error occurs.

It might be hard to understand why one would put JSON in a query string, but in a situation where the existing system is being recreated, it was necessary to ensure backward compatibility.

I initially thought encoding was unnecessary because I knew that uriBuilder internally performs encoding. However, an error occurs in the expand logic of this.uriComponentsBuilder.build().expand(uriVars).

public class DefaultUriBuilderFactory implements UriBuilderFactory {

@Override
public URI build(Object... uriVars) {
    ... 

    UriComponents uric = this.uriComponentsBuilder.build().expand(uriVars);

    ...
}

The cause of the error was the processing according to the URI template. It recognized the curly braces in the JSON as a template for parsing, and although it looked for a value corresponding to the key 'name,' the error occurred because there was no value.

In other words, the expected template was in the form of /test?key={name}, but since the JSON format did not match this, it was determined that there was no value.

The error message is as follows.

msg: Not enough variable values available to expand '"name"'
java.lang.IllegalArgumentException: Not enough variable values available to expand '"name"'

at org.springframework.web.util.UriComponents$VarArgsTemplateVariables.getValue(UriComponents.java:370)
at org.springframework.web.util.HierarchicalUriComponents$QueryUriTemplateVariables.getValue(HierarchicalUriComponents.java:1093)
at org.springframework.web.util.UriComponents.expandUriComponent(UriComponents.java:263)
at org.springframework.web.util.HierarchicalUriComponents.lambda$expandQueryParams$5(HierarchicalUriComponents.java:456)
at java.base/java.util.Map.forEach(Map.java:713)
at org.springframework.web.util.HierarchicalUriComponents.expandQueryParams(HierarchicalUriComponents.java:452)
at org.springframework.web.util.HierarchicalUriComponents.expandInternal(HierarchicalUriComponents.java:441)
at org.springframework.web.util.HierarchicalUriComponents.expandInternal(HierarchicalUriComponents.java:53)
at org.springframework.web.util.UriComponents.expand(UriComponents.java:172)
at org.springframework.web.util.DefaultUriBuilderFactory$DefaultUriBuilder.build(DefaultUriBuilderFactory.java:403)
at org.springframework.web.util.DefaultUriBuilderFactory.expand(DefaultUriBuilderFactory.java:154)
at org.springframework.web.reactive.function.client.DefaultWebClient$DefaultRequestBodyUriSpec.uri(DefaultWebClient.java:194)

To solve this problem, the request was sent to the WebClient with double encoding (performing encoding on the JSON data in advance), and the server receiving this request had to perform decoding twice.

Whether combining the path and query string into a string and inserting it into the URI, or using uriBuilder, both methods result in the same error.

// path value is /test?jsonParam=%7B%0A%20%20%22name%22%3A%20%22ybs%22%0A%7D
// The jsonParam is the value obtained by encoding the below JSON once.
// {
//    "name": "ybs"
//  }
this.webClient
  .post()
  .uri(path)


// path value is /test
// query value is %7B%0A%20%20%22name%22%3A%20%22ybs%22%0A%7D
this.webClient
  .post()
  .uri(uriBuilder -> {
    return uriBuilder
      .path(path)
      .query(query)
      .build();

Upon looking for other solutions, it was found that by constructing it as below, it is possible to avoid the problematic expand logic, and thus no error occurs.

However, using this method slightly changes the encoded JSON result.

// method1
this.webClient
  .post()
  .uri(uriBuilder -> {
    URI uri = UriComponentsBuilder.fromUriString(path)
      .queryParam("query", query)
      .build()
      .toUri();
    return uri;
  }
)

// method2. Instead of using the uriBuilder lambda, you can directly put a URI object.
URI uri = UriComponentsBuilder.fromUriString(path)
  .queryParam("query", query)
  .build()
  .toUri();

return this.webClient
  .post()
  .uri(uri)

When looking at the query string value of the created URI object, the colon (:) is not encoded. In other words, when you URL encode a colon (:), it becomes %3A, but if the encoding is done internally by the URI, the result does not apply the encoding.

1. %7B%0A%20%20%22name%22:%20%22ybs%22%0A%7D (The result encoded internally by the URI)
2. %7B%0A%20%20%22name%22%3A%20%22ybs%22%0A%7D (The result of encoding in advance)

In the RFC 3986 Uniform Resource Identifier (URI) specification, a colon (:) is allowed as a query character. In other words, a colon (:) is permitted as a character within a query.

query       = *( pchar / "/" / "?" )
—————
pchar         = unreserved / pct-encoded / sub-delims / ":" / "@"
unreserved  = ALPHA / DIGIT / "-" / "." / "_" / "~"
pct-encoded   = "%" HEXDIG HEXDIG
sub-delims  = "!" / "$" / "&" / "'" / "(" / ")"
                  / "*" / "+" / "," / ";" / "="

Of course, whether URL encoding (referred to as percent-encode in the RFC specification) is not necessary can be seen as a separate issue. The reserved characters defined below are a set of characters that, as part of the URI syntax, can distinguish different data within the URI.

reserved    = gen-delims / sub-delims
—————
gen-delims  = ":" / "/" / "?" / "#" / "[" / "]" / "@"
sub-delims  = "!" / "$" / "&" / "'" / "(" / ")"
                  / "*" / "+" / "," / ";" / "="

The RFC 3986 specification document states that reserved characters should be encoded, but it also mentions that encoding is not necessary if they are specifically allowed in the component corresponding to the query.

Ultimately, the interpretation of the component may vary, but I understand that it is acceptable not to encode the colon (:), without causing any issues.

https://www.rfc-editor.org/rfc/rfc3986#section-2.2

URI producing applications should percent-encode data octets that
correspond to characters in the reserved set unless these characters
are specifically allowed by the URI scheme to represent data in that
component.  If a reserved character is found in a URI component and
no delimiting role is known for that character, then it must be
interpreted as representing the data octet corresponding to that
character's encoding in US-ASCII.

jackson-dataformat-xml issues

yangbongsoo — Sat, 16 Mar 2024 15:29:51 +0000

I summarized the issues encountered while using jackson-dataformat-xml(2.12.4).

1. interface naming

'interface' is a reserved word in Java, making it a poor choice for an XML element name.

However, the existing code was using the 'interface' naming, and it was necessary to develop in accordance with the same 'interface' naming to maintain backward compatibility.

Therefore, I first attached the @XmlElement annotation to the field to be bound, specifying the name as 'interface' like below.



@XmlElement(name = "interface")
private String interFace;

Since the Lombok library was used, the setter was automatically created with the naming 'setInterFace'.



public void setInterFace(String interFace) {
    this.interFace = interFace;
}

However, the value of the XML 'interface' tag was not being bound correctly. Instead, changing the setter naming as follows resulted in successful binding. In other words, the 'f' was changed to lowercase.



public void setInterface(String interFace) {
    this.interFace = interFace;
}

To understand the exact reason why the binding failed, I conducted debugging.

In attempting to find 'interface' within _beanProperties, no result appears, leading to the prop becoming null.
Consequently, an error occurs in handleUnknownVanilla with com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException: Unrecognized field "interface".

Another solution to this problem is utilizing the _caseInsensitive attribute found in _beanProperties.

By adding the configuration objectMapper.configure(MapperFeature.ACCEPT_CASE_INSENSITIVE_PROPERTIES, true), it becomes possible to find in _beanProperties without distinguishing between uppercase and lowercase, thereby finding the properties correctly.

Additionally, using @JacksonXmlProperty instead of @XmlElement results in 'interface' being registered in _beanProperties, making it discoverable.



@JacksonXmlProperty(localName = "interface")
private String interFace;

However, We decided not to apply it in practice because adding the dependency on com.fasterxml.jackson.dataformat.xml.annotation directly, rather than relying on javax.xml.bind.annotation, was deemed not advisable.

2. list

If multiple XML tags of the same kind are input, they can be bound to a Java List object. However, if the same XML tags do not come in succession, the behavior differs. Let's look at the example code below.



package com.fasterxml.jackson.dataformat.xml.lists;

import java.util.List;

import javax.xml.bind.annotation.XmlElement;
import javax.xml.bind.annotation.XmlRootElement;

import com.fasterxml.jackson.annotation.JsonMerge;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.dataformat.xml.XmlMapper;
import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlCData;
import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlElementWrapper;
import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlProperty;
import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlRootElement;

public class ListTest {
    public static void main(String[] args) throws JsonProcessingException {
        String xml = ""
            + "<Data>"
            + "    <product>a</product>"
            + "    <product>b</product>"
            + "    <backUrl>www.naver.com</backUrl>"
            + "    <product>c</product>"
            + "    <product>d</product>"
            + "    <backUrl>www.naver.com</backUrl>"
            + "    <product>e</product>"
            + "</Data>";

        XmlMapper m = new XmlMapper();
        // m.setDefaultMergeable(true);
        Data data = m.readValue(xml, Data.class);
        System.out.println(data.getProduct());
    }
}

@JacksonXmlRootElement(localName = "data")
class Data {
    @JacksonXmlCData
    @JacksonXmlElementWrapper(useWrapping = false)
    @JacksonXmlProperty
    private List<String> product;

    @JacksonXmlCData
    @JacksonXmlElementWrapper(useWrapping = false)
    @JacksonXmlProperty
    private String backUrl;

    public List<String> getProduct() {
        return product;
    }

    public void setProduct(List<String> product) {
        this.product = product;
    }

    public String getBackUrl() {
        return backUrl;
    }

    public void setBackUrl(String backUrl) {
        this.backUrl = backUrl;
    }
}

Multiple 'product' tags are supposed to be bound to a List object, but when actually calling getProduct, only 'e' appears. Although the XML request format was not standard, proper List object binding occurred with the older jaxb-api(2.1), necessitating a solution for this issue within jackson-dataformat-xml as well.

The first thing we can do is modify the setProduct method as follows. Changing it to add the product instead of overwriting it will yield the desired result.



public void setProduct(List<String> product) {
    if (this.product == null) {
        this.product = product;
    } else {
        this.product.addAll(product);
    }
}

Another method exists as well. From jackson-dataformat-xml(2.9) onwards, you can use the provided setDefaultMergeable.

Result without setDefaultMergeable true setting: [e]
Result with setDefaultMergeable true setting: [a, b, c, d, e]

However, there's one point of caution when using setDefaultMergeable. If the getter for the product is implemented as below to defend against null, then since the initial this.product is null, the result of the getProduct method becomes EMPTY_LIST.



public List<String> getProduct() {
    return Objects.isNull(this.product) ? 
    Collections.emptyList() :
    Collections.unmodifiableList(this.product);
}

Internally, when merging, the getProduct method is called, but since it initially receives EMPTY_LIST as a result, addition doesn't occur. This leads to an error: org.springframework.http.converter.HttpMessageNotReadableException: JSON parse error: (was java.lang.UnsupportedOperationException); nested exception is com.fasterxml.jackson.databind.JsonMappingException: (was java.lang.UnsupportedOperationException), even when using @JsonMerge.

Therefore, either implement getProduct in its basic form or



public List<String> getProduct() {
    return this.product;
}

work as follows to prevent a null return.



public List<String> getProduct() {
    if (this.product == null) {
        this.product = new ArrayList<>();
    }
    return this.product;
}

Finally, XmlMapper is a subclass of ObjectMapper. Setting XmlMapper.setDefaultMergeable(true) is fine because it calls the setDefaultMergeable method implemented by the superclass ObjectMapper.

Deep dive into ObjectMapper

yangbongsoo — Sun, 10 Mar 2024 06:29:09 +0000

1. serialize/deserialize issue

When operating on an EDA (Event-Driven Architecture) basis, particular attention must be paid to the serialization/deserialization of messages. This is because the systems are distributed.

For example, consider the scenario where System1, as depicted below, modifies a message-related object. System1 is responsible for serializing and publishing the message. However, other organizational systems like System2 and System3 might have different deployment timings. If the changes are not reflected, mismatches in consumption rules can lead to failures. Therefore, it is crucial to manage the ObjectMapper effectively.

1-1. FAIL_ON_UNKNOWN_PROPERTIES

The FAIL_ON_UNKNOWN_PROPERTIES option always requires careful attention.

The FAIL_ON_UNKNOWN_PROPERTIES option is an attribute that determines whether deserialization should fail when there is no matching field (unknown property) during deserialization.
cf) When a new field is added to an object for serialization and deployed, but the consuming systems (where the object is deserialized) do not have that field reflected.

If this attribute is set to true, a JsonMappingException exception occurs, leading to a failure. To prevent this, it should be set to false (to ignore unknown properties) or the object being deserialized should be additionally annotated with @JsonIgnoreProperties.

1-2. Spring Boot FAIL_ON_UNKNOWN_PROPERTIES

When using Spring Boot, the FAIL_ON_UNKNOWN_PROPERTIES option is adjusted to FALSE by default in Jackson2ObjectMapperFactoryBean.

However, if a new ObjectMapper is created directly, it does not go through Jackson2ObjectMapperFactoryBean, thus FAIL_ON_UNKNOWN_PROPERTIES remains true (the default value).

Therefore, it needs to be disabled as shown in the code below.

private static ObjectMapper objectMapper() {
    ObjectMapper objectMapper = JsonMapper.builder()
        .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
        .build()
}

1-3. codehaus and fasterxml

There are services that mix the use of org.codehaus.jackson(Jackson 1.x) and com.fasterxml.jackson(Jackson 2.x). Given the service has a long history, it has multiple ObjectMappers. Therefore, although there's a desire to add options to the ObjectMapper, predicting the scope of impact is challenging, making modifications difficult. In other words, there is a significant risk when ObjectMapper management is neglected. If predicting the scope of impact becomes difficult upon changing ObjectMapper settings, it's safer to add the @JsonIgnoreProperties(ignoreUnknown = true) annotation to all classes.

In practice, there have been instances where problems occurred. Despite importing com.fasterxml.jackson.annotation.JsonIgnoreProperties and adding the @JsonIgnoreProperties annotation, the ObjectMapper used for deserialization was from codehaus. Consequently, a Could not read JSON: Unrecognized field error occurred. Mistakes during import can lead to issues, and there's a high chance that these might be overlooked even during PR reviews. In cases where services mix libraries, it's advisable to standardize on codehaus since Jackson 2 also supports it, making it a safe choice. Later, when phasing out Jackson 1, it's better to switch to fasterxml in a unified manner.

2. Understanding the Principles of Deserialization

If Jackson decides to deserialize into the Sample class type, it first looks for a no-argument constructor in the Sample class. If one exists, it uses that constructor to create a Sample instance and then looks for setters. It then checks for setters with the same name as the keys in the JSON string and injects the values accordingly.

public static class Sample {
    private String name;
    private String address;

    public Sample() { }

    public void setName(String name) {
        this.name = name;
    }

    public void setAddress(String address) {
        this.address = address;
    }
}

Let's look at the other case. JSON delivers data such as name, address, and gender as follows

name : "ybs",
address : "seoul",
gender : "male"

but in the Sample object being deserialized into, there is no setter for the gender field. In this case, the name and address are injected through their setters, and since there is no setter for gender, it checks for a field with the same name. If found, it injects the value using reflection. If not found, it either ignores it or throws an exception, depending on the FAIL_ON_UNKNOWN_PROPERTIES setting.

// 1. setter
// 2. reflection
public static class Sample {
    private String name;
    private String address;
    private String gender;

    public Sample() { }

    public void setName(String name) {
        this.name = name;
    }

    public void setAddress(String address) {
        this.address = address;
    }
}

Next, if there is an all-argument constructor, it injects the values through this constructor. However, simply adding the constructor is not enough; it must be indicated with @JsonCreator and @JsonProperty annotations.

// 1. constructor
// 2. setter
// 3. reflection
public static class Sample {
    private String name;
    private String address;
    private String gender;

    @JsonCreator
    public Sample(
        @JsonProperty("name") String name,
        @JsonProperty("address") String address,
        @JsonProperty("gender") String gender
    ) {
        this.name = name;
        this.address = address;
        this.gender = gender;
    }

    public void setName(String name) {
        this.name = name;
    }

    public void setAddress(String address) {
        this.address = address;
    }
}

However, since the above method can be cumbersome, there is an alternative approach that involves using the @ConstructorProperties annotation, as follows.

public static class Sample {
    private final String name;
    private final String address;
    private final String gender;

    @ConstructorProperties({"name", "address", "gender"})
    public Sample(String name, String address, String gender) {
        this.name = name;
        this.address = address;
        this.gender = gender;
    }
}

How does deserialization work with inheritance? First, it injects values through the constructor, but there's no field for the parent. That means it doesn't get set through the constructor. Next, it looks for a setter, but if there's no setter available, finally, it resorts to reflection to inject the field, since the field exists

// 1. constructor
// 2. setter
// 3. reflection
public static abstract class Parent {
    private String parent;
}

public static class Sample extends Parent {

    @ConstructorProperties({"name", "address"})
    public Sample(String name, String address) {
        this.name = name;
        this.address = address;
    }

    public void setName(String name) {
        this.name = name;
    }

    public void setAddress(String address) {
        this.address = address;
    }    
}

However, since using setters is preferable to reflection, setters can be created as follows

public static abstract class Parent {
    private String parent;

    void setParent(String parent) {
        this.parent = parent;
    }
}

For Collection types, there's an additional step (getter). As in the code below, there might not be a setter, but a getter could exist. If it's a Collection type, Jackson calls the getter. If the list returned by the getter is null, it proceeds to the next step. However, if the list is not null and a list is returned, Jackson adds to this list.

// 1. constructor
// 2. setter
// 3. getter !!
// 4. reflection
public static class Sample {
    private List<String> samples = new ArrayList<>();

    public List<String> getSamples() {
        return samples;
    }
}

There's a caveat to be aware of. You should not use the List.of() operator as shown below because it results in ImmutableCollections.emptyList(), which is unmodifiable. Therefore, an error occurs the moment Jackson attempts to add to it.

public static class Sample {
    private List<String> samples = List.of();

    public List<String> getSamples() {
        return samples;
    }
}

Similarly, if the getter wraps the list with unmodifiableList, an error occurs the moment Jackson attempts to add to it. Therefore, as a solution, one should either handle it with @JsonIgnore or avoid using unmodifiableList.

public static class Sample {
    private List<String> samples = new ArrayList<>();

    // Add @JsonIgnore or remove unmodifiableList()
    public List<String> getSamples() {
        return Collections.unmodifiableList(samples);
    }
}

3. Deserialization with polymorphism

When you specify the necessary concrete class (Sample) while using readValue with ObjectMapper, it automatically deserializes into the Sample type.

Sample sample = OBJECT_MAPPER.readValue(json, Sample.class);

When you need to deserialize using an interface, it can occur when the framework automatically handles the deserialization for you. For example, consider the scenario where there are AEvent, BEvent, and CEvent classes that implement the Event interface. The raiseEvent method receives an argument of the Event type, but you want it to be deserialized into the appropriate concrete class based on the situation.

public static interface Event {
  ...
}

public static class AEvent implements Event { ... }
public static class BEvent implements Event { ... }
public static class CEvent implements Event { ... }

... 

protected void raiseEvent(Event event) {
    delegate.raiseEvent(event);
}

Of course, we can create a class capable of holding all messages and deserialize into it. However, this approach requires understanding which fields are used and which are not, making it challenging to manage. The prevalence of nullable cases makes it convenient initially but difficult to maintain over time.

To deserialize into the appropriate concrete class even when dealing with an interface type, there needs to be something that can identify the correct class. This is where @JsonTypeInfo comes in handy. By attaching this annotation, the type information is embedded during serialization, indicating which class to deserialize into.

@JsonTypeInfo(use = Id.NAME, include = As.WRAPPER_OBJECT)
private List<Event> events = new ArrayList<>();

In other words, when serializing an interface type, the type name is also serialized alongside the data to ensure that the type can be identified upon deserialization. This is done by embedding an id value, which is used to select the correct class to deserialize into. If the strategy is WRAPPER_OBJECT, the data is wrapped, resulting in an additional layer of depth. This can be observed in the JSON structure, where, for example, AEvent would be used as a key, and the actual data would be nested within an additional layer beneath it.

{
  "events":
    [
      {
        "AEvent":{"sourceVersion":100,"id":1}
      }

        ...
    ]
}

Changing @JsonTypeInfo settings requires caution, as errors can occur if the deserialization side is not properly aligned. Since Id.NAME is used with @JsonTypeInfo, serialization passes only the class name, so changes in the package do not affect it. However, if Id.CLASS is used, it serializes the fully-qualified Java class name, including the package name, so any change in the package affects deserialization.

{
  "events":
    [
      {
        "com.toy.AEvent":{"sourceVersion":100,"id":1}
      }

        ...
    ]
}

And if PROPERTY is used instead of WRAPPER_OBJECT, the type information is passed using the @type key."

{
  "events":
    [
      {
        "@type":"AEvent"
        "sourceVersion":100,
        "id":1
      }

        ...
    ]
}

Next, the @JsonSubTypes annotation, which matches @JsonTypeInfo, is specified separately on the setter. This registers the mapping information to determine the restoration type based on the serialized name when deserializing an interface type.

@JsonSubTypes({
    @Type(value = AEvent.class, name = "AEvent"),
    @Type(value = BEvent.class, name = "BEvent"),
    @Type(value = CEvent.class, name = "CEvent"),
})
protected void setEvents(List<Event> Events) {
    this.Events = Events;
}

cf) Caution: When adding a new Event type, it must be added here to ensure proper deserialization.

If DEvent is created but not added to JsonSubTypes, the following error occurs:

com.fasterxml.jackson.databind.exc.InvalidTypeIdException:
Could not resolve type id 'DEvent' as a subtype of `com.toy.Event`: known type ids = [AEvent, BEvent, CEvent] (for POJO property 'events')
at [Source: (String)"{"events":[{"AEvent":{"type":"AEvent","sourceVersion":100,"id":1}},{"BEvent":{"type":"BEvent","sourceVersion":200,"id":2}},{"CEvent":{"type":"CEvent","sourceVersion":300,"id":3}},{"DEvent":{"type":"DEvent","sourceVersion":400,"id":4}}]}"; line: 1, column: 181] (through reference chain: com.toy.Process["events"]->java.util.ArrayList[3])

4. Other Options in ObjectMapper

First, we disable the ALLOW_FINAL_FIELDS_AS_MUTATORS setting.

.disable(MapperFeature.ALLOW_FINAL_FIELDS_AS_MUTATORS)

"In the code below, the final field 'name' was initialized at the time of declaration, making it immutable thereafter. However, during deserialization, Jackson can use reflection to modify its value. To prevent this, we disable the ALLOW_FINAL_FIELDS_AS_MUTATORS setting.

public static class Sample {
    private final String name = "ybs";
}

We also disable the FAIL_ON_EMPTY_BEANS setting.

.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS)

When deserializing the Sample class, Jackson basically throws an error if there are no fields defined in the class. At least one field must be present. Therefore, we disable the FAIL_ON_EMPTY_BEANS setting.

public static class Sample {
}

The situation where this option is necessary is as follows: when data of types A, B, and C is received from a Kafka topic, but you want to ignore type C. While filtering can be done at the front end using Kafka header values, there are cases where you need to deserialize the data to determine its type. However, since there's no need to define fields for type C that you won't use, you can create a shell class and skip it when necessary.

Type Cache

The LRUMap is used as a cache for the TypeFactory, allowing Jackson to internally cache information about the types during serialization and deserialization. This means that when searching for a required type, Jackson first checks the cache, retrieves it if found, and uses it.

return objectMapper.setTypeFactory(
    objectMapper.getTypeFactory()
        .withCache((LookupCache<Object, JavaType>)new LRUMap<Object, JavaType>(5120, 5120))
);

The default cache size is 200.

protected TypeFactory(LookupCache<Object,JavaType> typeCache, TypeParser p,
                      TypeModifier[] mods, ClassLoader classLoader)
{
    if (typeCache == null) {
        // initialEntries : 16
        // maxEntries : 200
        typeCache = new LRUMap<>(16, 200);
    }
}

When the cache becomes full, if a new type needs to be added, Jackson internally clears the LRUMap within a synchronized block and starts over from scratch. Since 200 slots can fill up quickly, the cache size was increased to 5120.

package com.fasterxml.jackson.databind.util;

public class LRUMap<K,V>
    implements LookupCache<K,V>, // since 2.12
        java.io.Serializable
{
    @Override
    public V put(K key, V value) {
        if (_map.size() >= _maxEntries) {
            // double-locking, yes, but safe here; trying to avoid "clear storms"
            synchronized (this) {
                if (_map.size() >= _maxEntries) {
                    clear();
                }
            }
        }
        return _map.put(key, value);
    }

    ...
}

Lastly, let's examine exactly what is being cached. In the LookupCache<Object, JavaType>, the key is an Object, and the value is a JavaType. JavaType (com.fasterxml.jackson.databind.JavaType) contains information about the type corresponding to the Object (key), including super class type, super interface type, and so on.

When deserializing the following JSON string into a YBS object, the items field of the YBS class is of type List interface, but the actual concrete type created is ArrayList.

// json = "{\"items\":[\"item1\",\"item2\"]}";

@Getter
@Setter
public class YBS {

    private List<String> items;

    @ConstructorProperties({"items"})
    public YBS(List<String> items) {
        this.items = items;
    }
}

As a result, JavaType is retrieved and cached for all objects associated with ArrayList.

public class ArrayList<E> extends AbstractList<E>
        implements List<E>, RandomAccess, Cloneable, java.io.Serializable { ... }

Ultimately, it can be observed that various objects are cached for a single List field.

Message Compensation Processing Due to DB Failure in EDA

yangbongsoo — Sun, 25 Feb 2024 08:24:18 +0000

Suddenly, all application servers failed to establish a network connection with the DB server. It's a situation of complete system failure. Let's explore what issues arise with the orderSheet server I am responsible for when there is a problem with the DB and why compensation processing is necessary.

// response message
Could not open JDBC Connection for transaction;
nested exception is java.sql.SQLException: Cannot get a connection, 
pool error Timeout waiting for idle object

When a user clicks the payment button on the orderSheet screen, the payment proceeds after passing through a validation check, and ultimately, the order completion process is carried out.

If a DB issue occurs before the payment is executed, the user may experience inconvenience, but there is no requirement for us to provide data compensation.

However, if a DB issue arises after the payment has been completed, we must perform compensation processing, such as canceling the payment.

The orderSheet server was designed with an EDA, but it communicated directly with the order/payment server via an API. If the DB server goes down, the order completion request (number 2) fails, followed by an attempt to perform the order failure event handling task (number 4), which also ends up failing (number 6).

cf) As of February 2024, the communication in this segment has also been switched to a Kafka publish/subscribe structure.

Now let's go into detail. When a user clicks the payment button on the orderSheet screen, the orderSheet server proceeds with validation and payment, and then receives an order completion event to start the final process.

@Component
@RequiredArgsConstructor
public static class CommandMessageHandler {
  // number 1
  public void onCompleteOrder(CompleteOrder completeOrder) {
    onCompleteOrder(completeOrder);
  }
}

If the API response fails, it publishes the onCompleteOrderFailed event and sends it to number 4.

public void completeOrder(CompleteOrder completeOrder) { 
  CompleteOrderRequestDto dto = toCompleteOrderRequestDto(completeOrder);

  try {    
    // number 2
    apiClient.completeOrder(dto).block();
  } catch (Exception ex) {
    // publish onCompleteOrderFailed event(go to number 4)
  }
}

Since the DB is down, the order failure processing request (number 5) also fails. However, it merely logs the error in the catch block and ends the process. In other words, it throws an exception to the KafkaListener.

@Component
@RequiredArgsConstructor
public static class MessageMessageHandler {
  // number 4
  public void onCompleteOrderFailed(CompleteOrderFailed completeOrderFailed) {
    onCompleteOrderFailed(completeOrderFailed);
  }
}


public void onCompleteOrderFailed(completeOrderFailed) {
  try {
    // number 5
    apiClient.rollbackForConsistence(completeOrderFailed);
  } catch(Exception ex) {
    log.error("error!!", ex)
  }
}

In springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder, the DefaultAfterRollbackProcessor is used as the afterRollbackProcessor. When creating the DefaultAfterRollbackProcessor object, a BackOff is also provided. If you look at createBackOff, you'll see that the maxAttempts in ConsumerProperties is set to 3.

maxAttempts signifies the number of attempts to process a message in case of failure (including the first attempt). It is a configuration of the RetryTemplate provided by the framework. Since we were using the default setting, it attempts to consume a total of 3 times.

messageListenerContainer.setAfterRollbackProcessor(new DefaultAfterRollbackProcessor<>(
  (record, exception) -> {
    MessagingException payload =
        new MessagingException(((RecordMessageConverter) messageConverter)
              .toMessage(record, null, null, null),
              "Transaction rollback limit exceeded", exception);
    try {
      errorInfrastructure.getErrorChannel()
          .send(new ErrorMessage(
              payload,
              Collections.singletonMap(
                IntegrationMessageHeaderAccessor.SOURCE_DATA,
                record
              )
            )
          );
    }
    catch (Exception e) {
      /*
      * When there is no DLQ, the FinalRethrowingErrorMessageHandler will re-throw
      * the payload; that will subvert the recovery and cause a re-seek of the failed
      * record, so we ignore that here.
      */
      if (!e.equals(payload)) {
        throw e;
      }
    }
  }, createBackOff(extendedConsumerProperties),
  new KafkaTemplate<>(transMan.getProducerFactory()),
  extendedConsumerProperties.getExtension().isTxCommitRecovered()));

Since the attempt to consume the event was retried 3 times but all attempts failed, ultimately, the following error message is received.

Transaction rollback limit exceeded; nested exception is 
org.springframework.kafka.listener.ListenerExecutionFailedException

Now let's look into compensation processing.
After the DB server is normalized, one way to correct the data is by manually republishing the onCompleteOrderFailed event, allowing the order failure process (number 4) to be reattempted.

This will proceed with the order failure handling, and the subsequent rollback process will operate to address the tangled data.

To manually republish the onCompleteOrderFailed event, it is necessary to first know the topic, partition, and offset information.

The topic can be found in the application property settings, and the partition and offset can be located in the error logs (the offset where the producer published the order failure event after an API error occurred in number 3).

Once the offset is identified, the next step is to find the message at that offset. We located it by searching through our company's operational tools. Then, we copied the message value, generated a new key for it, and manually republished the event.

It's regrettable that I cannot provide detailed explanations on this part due to it being a personal blog.

Finally, what happens when a DB outage occurs in a Kafka pub/sub structure, rather than through API communication? We won't receive an API error response, but command/event objects will remain stagnant at a certain state.

We collect objects that continue to be stagnant even after a certain period and proceed with the compensation processing.

Be careful of NullPointerException in EDA

yangbongsoo — Sat, 17 Feb 2024 15:22:24 +0000

In a Spring application, if messages are not exchanged directly but through Kafka based on an event-driven approach, special attention must be paid to avoiding NullPointerException.

Let's look at the code below.
In onStartOrder, the order process begins and publishes a point reservation command message to Kafka. Then, in onReservePoint, it receives and processes the point reservation command message.

The point reservation command message contains the pointReserveDto, but with a specification change, a new field became necessary. Thus, a newObject was created and additionally passed, and the modification was made in onReservePoint, which performs the point reservation, to retrieve and use it.

Here arises a problem. Existing point reservation command messages in Kafka do not contain the newObject in their pointReserveDto. Although the Spring application has been deployed with a new version, it still consumes old version messages and executes the code pointReserveDto.getNewObject().getData(), leading to a NullPointerException.

I have illustrated the entire process to provide a detailed explanation through code. It's beneficial to view the workflow and code together.
cf) The code for steps 5 and 6 has been omitted, as discussing persistence would make the explanation overly lengthy.

The starting point is the kafkaListener. When it receives a StartOrder command message from Kafka, it delegates the task to the orderService.

@Configuration
@RequiredArgsConstructor
public class KafkaListener {

    @Component
    @RequiredArgsConstructor
    public static class KafkaCommandHandler {
        private final OrderService orderService;

        public void onStartOrder(StartOrder startOrder) {
            this.orderService.startOrder(startOrder);
        }
    }
}

In orderService.startOrder, a PointReserveDto is created, and due to a specification change, a NewObject is also created and included. It is then passed to the orderProcessor.

@Service
@RequiredArgsConstructor
public class OrderService {

    private final OrderProcessor orderProcessor;

    public void startOrder(StartOrder startOrder) {
        String orderId = startOrder.getOrderId();

        // create NewObject due to change the specification
        PointReserveDto pointReserveDto = PointReserveDto.builder()
            .newObject(
                NewObject.builder()
                    .data("newData")
                    .build()
            )
            .build();

        orderProcessor.processStartOrder(orderId, pointReserveDto);
    }
}

In orderProcessor, business logic is executed. First, it retrieves the Order object from the database using the orderId received from the StartOrder command. Then, utilizing the Order object, it carries out the tasks related to starting the order. Upon completion, it calls order.applyOrderStarted to publish an event indicating that the StartOrder has finished.

@Component
public class OrderProcessor {
    public void processStartOrder(String orderId, PointReserveDto pointReserveDto) {
        Order order = getOrderById(orderId);

        // Do StartOrder process

        order.applyOrderStarted(pointReserveDto);
    }

    // In reality, it is necessary to query the database to retrieve the Order object.
    private Order getOrderById(String orderId) {
        return Order.builder().id(orderId).build();
    }
}

cf) The main purpose is to explain the process leading to a NullPointerException, so we'll skip details, but it's worth noting that the structure of issuing events and commands from Order is a key point in an EDA.

Next, let's look at the following code. order.applyOrderStarted takes the received pointReserveDto, encapsulates it into an OrderStarted event, and adds it to the pendingEvents list. Lastly, through the apply method, it fills the Order object's pointReserveDto field.

Subsequently, it creates a ReservePoint command, encapsulates the pointReserveDto within it, and similarly adds it to the pendingCommands list. The remaining tasks involve publishing the messages in pendingEvents and pendingCommands to Kafka and saving the order in the database. This is the content for step 5 and 6, and the related code is omitted.

@Getter
@Builder
public class Order {
    String id;

    @Nullable
    PointReserveDto pointReserveDto;

    List<Event> pendingEvents = new ArrayList<>();
    List<Command> pendingCommands = new ArrayList<>();

    public void applyOrderStarted(PointReserveDto pointReserveDto) {
        this.pendAndApplyEvent(
            OrderStarted.builder()
                .orderId(this.getId())
                .pointReserveDto(pointReserveDto)
                .build(),
            this::apply
        );

        this.pendingCommands.add(
            ReservePoint.builder()
                .orderId(this.getId())
                .pointReserveDto(this.pointReserveDto)
                .build()
        );
    }

    private <T extends Event> void pendAndApplyEvent(T event, Consumer<T> apply) {
        this.pendingEvents.add(event);
        apply.accept(event);
    }

    private void apply(OrderStarted orderStarted) {
        this.pointReserveDto = orderStarted.getPointReserveDto();
    }
}

Because the ReservePoint command was published above, the KafkaCommandHandler consumes it again through onReservePoint.

@Configuration
@RequiredArgsConstructor
public class KafkaListener {

    @Component
    @RequiredArgsConstructor
    public static class KafkaCommandHandler {
        private final OrderService orderService;

        public void onReservePoint(ReservePoint reservePoint) {
            this.orderService.reservePoint(reservePoint);
        }
    }
}

In orderProcessor, attempting to extract the newObject.data value from the received pointReserveDto leads to encountering a NullPointerException.

@Component
public class OrderProcessor {
    public void processReservePoint(
        String orderId,
        PointReserveDto pointReserveDto
    ) {
        Order order = getOrderById(orderId);

        // Do ReservePoint process

        // Fixme: NPE !! pointReserveSource.getNewObject() is Null
        String newData = pointReserveDto.getNewObject().getData();
    }

    // In reality, it is necessary to query the database to retrieve the Order object.
    private Order getOrderById(String orderId) {
        return Order.builder().id(orderId).build();
    }
}

Webclient timeout and connection pool Strategy

yangbongsoo — Sun, 11 Feb 2024 15:34:37 +0000

1. Webclient timeout

Let's look at the code below. There are many timeout options.



new ReactorClientHttpConnector(
    reactorResourceFactory,
    httpClient -> httpClient
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
        .doOnConnected(connection ->
            connection.addHandlerLast(new ReadTimeoutHandler(5)
            ).addHandlerLast(new WriteTimeoutHandler(5))
        ).responseTimeout(Duration.ofSeconds(5)) // added since version 0.9.11
);

ChannelOption.CONNECT_TIMEOUT_MILLIS is a waiting time for establishing a connection with the server and It is httpClient level.

responseTimeout is a timeout purely for HTTP request/response time.

Using ReadTimeoutHandler/WriteTimeoutHandler as a substitute for responseTimeout is not appropriate.
cf) responseTimeout was added starting from version 0.9.11, so alternative timeouts had to be utilized prior to that.

Especially since it applies at the TCP level, it also affects during the TLS handshake. Therefore, due to the associated encryption operations, it may take longer than a typical HTTP response. Consequently, the timeout should be set higher than that for pure HTTP responses.

Furthermore, the ReadTimeoutHandler/WriteTimeoutHandler handlers operate even when an HTTP request is not performed. For example, the handlers can close a connection in the connection pool, even if it could be swiftly used by another request later on.

In summary, ReadTimeoutHandler is not directly related to HTTP. It should be understood as a standard Netty handler that checks the time between different read operations.

The timeout() method of reactive streams is also insufficient for use as a responseTimeout. It covers not only the time the client takes to receive a response but also includes the operations of obtaining a connection from the connection pool and creating new connections within the reactive stream (including the TLS handshake process).

Therefore, responseTimeout must always be greater than the connection timeout plus the time to obtain a connection from the connection pool.

In other words, the response timeout limits only the pure HTTP request/response time, without considering the time to close idle connections or establish new connections.

Reference: https://github.com/reactor/reactor-netty/issues/1159

2. Webclient connection pool



ConnectionProvider provider = ConnectionProvider.builder("ybs-pool")
    .maxConnections(500)
    .pendingAcquireTimeout(Duration.ofMillis(0))
    .pendingAcquireMaxCount(-1)
    .maxIdleTime(Duration.ofMillis(8000L))
    .maxLifeTime(Duration.ofMillis(8000L))
    .build();

maxLifeTime : "The maximum lifetime of a connection that can exist in the connection pool
maxIdleTime : The duration for which idle connections are maintained in the connection pool
pendingAcquireMaxCount : The maximum time to wait for obtaining a connection from the connection pool (inserting -1 means no timeout)

Caution: Depending on the target Tomcat server's connectionTimeout and keepAliveTimeout settings, it may cause the connection to close.

Therefore, it's advisable to set the maxIdleTime on the client side with the Tomcat keepAliveTimeout in mind.

In other words, it is better for the maxIdleTime to be shorter than the keepAliveTimeout. Otherwise, Reactor Netty can receive a close event at any time 'between obtaining a connection from the pool and sending the actual request'.

To address this issue, a LIFO (Last In, First Out) strategy, which always uses the most recently used connection, can be employed.
Reference: https://github.com/reactor/reactor-netty/issues/1092#issuecomment-648651826

Many issues arise due to connections being closed because of timeouts. This has led to the ability to switch the pool's release strategy. FIFO is the default, and LIFO was added starting from version 0.9.5.

cf) FIFO stands for First In, First Out, a common example being a queue. LIFO stands for Last In, First Out, with a stack being an example.

LIFO + max idle timeout operates as follows.

When a connection is obtained, the most recent one is used.
If it reaches the max idle timeout, this connection will be closed, and since this connection was the most recently used, it means that the other remaining ones (not active) in the pool will also be closed. A new connection is created and used for the request.
If the connection is closed by the remote peer (between acquisition and actual use), a 'connection reset by peer' message will be sent, and we will retry the request. Since this connection was the most recently used and was closed by the remote peer, all the other remaining ones (not active) in the pool will also be closed. Therefore, a new connection is created and used for the second attempt.

FIFO always obtains the oldest connection. Therefore, just because the oldest connection has reached the max idle time does not mean the next connection will also hit the max idle time.

3. reactor.netty.http.client.PrematureCloseException: Connection prematurely closed BEFORE response

I'll introduce a case where a PrematureCloseException occurred.

The structure involved three servers: A -> B -> C

A Server (making polling requests every 3 seconds)
B Server (making requests to C Server using WebClient)
C Server (configured with an Nginx keepalive timeout of 3 seconds)

In this setup, if the request from B to C takes slightly longer than 3 seconds, C Server times out and terminates the connection (sending an RST), resulting in a PrematureCloseException error.

This issue was resolved by changing the keepAlive setting of reactor.netty.http.client.HttpClient to false.

The PrematureCloseException occurs due to the TCP RST flag. Here's more detail about the TCP RST flag:

RST : The Reset flag indicates that the connection should be rest, and must be sent if a segment is received which is apparently not for the current connection. On receipt of a segment with the RST bit set, the receiving station will immediately aobrt the connection. Where a connection is aborted, all data in transit is considered lost, and all buffers allocated to that connection are released.

There are various scenarios where a TCP RST segment can occur, meaning the causes of a PrematureCloseException can be diverse.

Scenario1: Sending Duplicate SYNs

TCP A sends a SYN with a sequence number of 200, but the transmission of the segment is delayed.
TCP B receives the delayed SYN segment with sequence number 90(line 3).
Since it's the first SYN received by TCP B, it sends an ACK, indicating that it will use an initial sequence number of 500. However, TCP A detects that the ACK field sent by TCP B is incorrect, thinking its SYN segment hasn't reached the destination, and thus sends a Reset to refuse the segment(line 5).
For credibility of the segment, TCP A uses a sequence number of 91 in this reset segment(line 5).
Upon receiving the segment with the Reset flag set, TCP B re-enters the LISTEN state(line 5).
TCP B responds with an ACK in a normal manner using its new sequence number(line 7).
TCP A confirms that the connection has been established with the ISN(Initial Sequence Number) in line 8.

Scenario2: When one side crashes and is restarted(error recovery mechanism)

TCP A and TCP B are operating normally when TCP B sends a segment, but then a crash occurs at TCP A.
TCP A closes and sends a SYN again for a three-way handshake.
TCP B believes it is synchronized (thinks the connection is still established). Thus, upon receiving the SYN segment, it checks the sequence number and realizes there is a problem (line 3).
TCP B sends back an ACK indicating it was expecting sequence number 150 (line 4).
TCP A thinks the received segment does not match what it sent (as it sent a SYN for the first step of the three-way handshake) and sends a Reset segment (line 5).
TCP B is disrupted and closes.
TCP A can now attempt to connect again with a three-way handshake (line 7).

Scenario3:

TCP A experiences a crash.
However, TCP B, not recognizing the crash event at TCP A, sends a segment containing data.
Since TCP A is unaware of such a connection's existence upon receiving this segment, it responds with a Reset.

Scenario4:

Scenario 4 starts with both parties in the LISTEN state.
At this point, the duplicate SYN issue seen in Scenario 1 occurs, and TCP A sends a Reset.
TCP B then returns to the LISTEN state.

Reference: TCP/IP The Ultimate Protocol Guide
Reference: Effective tcp/ip programming

How to handle a Map using Java stream

yangbongsoo — Sun, 04 Feb 2024 16:18:41 +0000

Let's look at the code below. I create a Product class that contains a list of Items.

@Value
@Builder
public class Product {
    long productId;

    List<Item> items;

    @Value
    @Builder
    public static class Item {
        long itemId;
    }
}

And then I get the product list using an external API. In order to handle item objects, I use a nested for loop.

I make itemsByItemId Map object. Key is itemId and value is item object.

public class Main {
    public static void main(String[] args) {
        // get product list using an external api
        ApiClient apiClient = new ApiClient();
        List<Product> products = apiClient.getProducts().collectList().block();

        Map<Long, Item> itemsByItemId = new HashMap<>();
        for (Product product : products) {
            for (Item item : product.getItems()) {
                itemsByItemId.put(item.getItemId(), item);
            }
        }

        System.out.println(itemsByItemId);
    }
}

However I want to use java stream api. Let's look at the code below. It looks complicated at a glance.

For each item, we create and merge Maps using Map.of, and since we need to merge again at the product level, we incorporate flatMap(map -> map.entrySet().stream()) and collect(Collectors.toMap) twice each.

public class Main {
  public static void main(String[] args) {
    // get product list using an external api
    ApiClient apiClient = new ApiClient();
    List<Product> products = apiClient.getProducts().collectList().block();

    Map<Long, Item> itemsByItemId = products.stream() // Stream<Product>
      .map(product ->
        product.getItems().stream() // Stream<Item>
          .map(item -> 
            Map.of(item.getItemId(), item)) // Stream<Map<Long, Item>>
            .flatMap(map -> map.entrySet().stream()) // Stream<Entry<Long, Item>>
            .collect(Collectors.toMap(
              Entry::getKey,
              Entry::getValue,
              (item1, item2) -> item2
          )
        )
      ) // Stream<Map<Long, Item>>
      .flatMap(map -> map.entrySet().stream()) // Stream<Entry<Long, Item>>
      .collect(Collectors.toMap(
        Entry::getKey,
        Entry::getValue,
        (item1, item2) -> item2
      )
    );

    System.out.println(itemsByItemId);
  }
}

This is a better solution code. I create item lists using flatMap. .flatMap(product -> product.getItems().stream()) And then, I only need to create them once with Map.of.

public class Main {
  public static void main(String[] args) {
    // get product list using an external api
    ApiClient apiClient = new ApiClient();
    List<Product> products = apiClient.getProducts().collectList().block();

    Map<Long, Item> itemsByItemId = products.stream() // Stream<Product>
      .flatMap(product -> product.getItems().stream()) // Stream<Item>
      .map(item -> Map.of(item.getItemId(), item)) // Stream<Map<Long, Item>>
      .flatMap(map -> map.entrySet().stream()) // Stream<Entry<Long, Item>>
      .collect(Collectors.toMap(
          Entry::getKey,
          Entry::getValue,
          (item1, item2) -> item2
        )
      );

    System.out.println(itemsByItemId);  
  }
}

However, in reality, since collect(Collectors.toMap) is used to create the Map, there's no need for me to manually work with Map.of. Ultimately, it resulted in much simpler code as shown below.

public class Main {
  public static void main(String[] args) {
    // get product list using an external api
    ApiClient apiClient = new ApiClient();
    List<Product> products = apiClient.getProducts().collectList().block();

    Map<Long, Item> itemsByItemId = products.stream() // Stream<Product>
      .flatMap(product -> product.getItems().stream()) // Stream<Item>
      .collect(Collectors.toMap(Item::getItemId, Function.identity()));

    System.out.println(itemsByItemId);  
  }
}

Java Mono.defer usecase

yangbongsoo — Sun, 04 Feb 2024 14:53:57 +0000

The definition of Mono.defer is described in detail in stack overflow comment. If we use Mono.defer, we can operate Mono.defer's wrapped code at the time of execution, not at the time of declaration. So I understood that It would be necessary to subscribe several times in one context.

However, even in a situation that was not the case, I met Mono.defer and couldn't understand at first.

Let's see below code. If we don't use Mono.defer, we have to declare separately Mono for switchIfEmpty. If there is a value in the cache, this.apiClient.getDataNoById(id).doOnSuccess(TODO) method is always executed even if switchIfEmpty is not executed.

// without using Mono.defer
public Mono<Data> requiredById(String id) {
  Mono<Long> alternate = this.apiClient.getDataNoById(id).doOnSuccess(TODO);

  return Mono.justOrEmpty(cache.getData(id))
    .switchIfEmpty(alternate)
    .flatMap(this::requiredOne)
    ...
}

If we use Mono.defer instead, we can make Mono.defer's wrapped code operate when switchIfEmpty runs.

// use Mono.defer
public Mono<Data> requiredById(String id) {

  return Mono.justOrEmpty(cache.getData(id))
    .switchIfEmpty(Mono.defer(() -> this.apiClient.getDataNoById(id).doOnSuccess(TODO)))
    .flatMap(this::requiredOne)
    ...
}

However, the getDataNoById method returns the Mono type and has not subscribed yet. Therefore, the actual webClient call does not occur. So I didn't understand why Mono.defer was necessary in there.

As a result, If switchIfEmpty supports the lambda expression, It doesn't have to use Mono.defer. But java doesn't support it, so they used Mono.defer.

cf) spring.web.reactive.dispatcherHandler also uses Mono.defer. Even though it is simply creating an Exception object and wrapping it with Mono.error, using Mono.defer seems to have written it as a way to create a clean code.

@Override
public Mono<Void> handle(ServerWebExchange exchange) {
  if (this.handlerMappings == null) {
    return createNotFoundError();
  }

  return Flux.fromIterable(this.handlerMappings)
    .concatMap(mapping -> mapping.getHandler(exchange))
    .next()
    .switchIfEmpty(createNotFoundError())
    .flatMap(handler -> invokeHandler(exchange, handler))
    .flatMap(result -> handleResult(exchange, result));
}

private <R> Mono<R> createNotFoundError() {
  return Mono.defer(() -> {
    Exception ex = new ResponseStatusException(HttpStatus.NOT_FOUND, "No matching handler");
    return Mono.error(ex);
  });
}