Sign in


i have set up a program to consume from a kafka topic.

it has been running fine, till ~ 25% of the messages (out of 2/3mil rows) are consumed, there would be a lot of reblance started happening.

i have tried to increase the `

max.poll.interval.ms

and reduce the `

max.poll.records

and to asynchronously process the messages. have tried to set up the `

onPartitionsAssigned

and `

onPartitionsRevoked

`method as well. however, the rebalance is still keep happening, and the program starts stucking at 25%.

after a lot of investigation and trials, turns out that was due to the memory…


TreeMap could be used to sort the map by value

Map<Character, Integer> sorted = new TreeMap<>((o1, o2) -> comparator.compare(map.get(o2), map.get(o1))); sorted.putAll(map);

however, there is a caveat there. if the values…


there are times even the application is requesting for parallel processing, it could stick to one single thread without rotation.

for example, the case, https://lwpro2.dev/2021/06/11/jdk-spliterator-with-parallel-processing/.

this could be the case…


I have created a fixed size Spliterator to split the collection into fixed size. https://github.com/openjdk/jdk/pull/2907 It's working as expected. With a collection of size greater than the threshold, it would trySplit and generate a new Spliterator. And if the stream is parallel, it would pass the spliterator to different threads.

however, when trying to group the elements by thread, below code is not always working

Map<String, List<Integer>> partition = new ForkJoinPool(10).submit( () ->StreamSupport.stream(new FixedSizeSpliterator<>(numbers.toArray(new Integer[0]), 5), true) .collect(Collectors.groupingBy(i-> Thread.currentThread().getName(),Collectors.toList()))).join(); partition.entrySet().stream().forEach(en ->log.warn("check key {} vs value {}", en.getKey(), en.getValue() ));

most of the time, it would equally divided the elements into…


similar to circuit breaker, spring retry is a mechanism to handle for intermittent service unavailability and fallback if that has been sustained.

my code has been working well

@Recover public boolean altXX(String date){ ...... return false; } @Retryable(value = DataXXXException .class, maxAttempts = 10, backoff = @Backoff(delay = 30_000)) public boolean xxCheck(String date) throws DataXXXException { //biz logic here log.error("capturing the error", e); throw new DataXXXException ("XX Data not yet ready"); } return true; }

it would log the error several times before either it return a true when the data is available or return false after 10 tries.

till…


when the forked out thread is running the task, if any exception happens, depends on the mechanism how the thread/threadpool is constructed, it could handle the exception differently.

for forkjoinpool, if the task is executed. when the task failed with exception, it will propagate the exception, till it’s either handled or if uncaught then captured by the `UncaughtExceptionHandler`. (this is called by JVM).

when the task is submitted though, the task is wrapped into AdaptedRunnableAction, which doesn't propagate the exception.

I think what JCP/JVM trying to do is, trust the developers know the difference between the call and handle the…

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store