Handle exceptions in Kafka stream application #18928
Conversation
| } | ||
|
|
||
| @Override | ||
| public void configure(Map<String, ?> configs) {} |
There was a problem hiding this comment.
I think this isn't using the Baeldung formatter - do you have access to the Baeldung IDE formatter profile? Similar elsewhere
There was a problem hiding this comment.
I found it from the earlier baeldung docs and made the changes.
| try { | ||
| return mapper.writeValueAsBytes(user); | ||
| } catch (JsonProcessingException ex) { | ||
| log.error("Error deserializing the user {} with exception {}", user, ex.getMessage()); |
There was a problem hiding this comment.
Best practice is to include the exception as the last parameter to the logger call too so the stack trace is preserved
There was a problem hiding this comment.
sure, i've updated to include it.
| try { | ||
| return mapper.readValue(bytes, User.class); | ||
| } catch (IOException e) { | ||
| log.error("Error deserializing the message {} for topic {}", bytes, topic); |
There was a problem hiding this comment.
Best practice is to include the exception as the last parameter to the logger call too so the stack trace is preserved
There was a problem hiding this comment.
sure, i've updated to include it.
| .untilAsserted(() -> { | ||
| ConsumerRecords<String, Long> records = consumer.poll(Duration.ofMillis(500)); | ||
| Map<String, Long> counts = StreamSupport.stream(records.spliterator(), false) | ||
| .collect(Collectors.toMap(ConsumerRecord::key, ConsumerRecord::value, (a, b) -> b)); |
There was a problem hiding this comment.
What's the reason for the (a, b) -> b part?
There was a problem hiding this comment.
This is to avoid any map error if duplicate present and so provided with a merge function for resolution. But can be removed as well in our case as duplicate not possible.
2 participants
No description provided.