WebFlux Error Handling

WebFlux Error Handling

• doOnError : 예외 발생시 특정행위를 수행한다.
• onErrorReturn : 예외 발생시 특정 값을 return 한다.
• onErrorResume : 예외 발생시 다른 Flux형태로 return 한다.
• onErrorContinue : 예외 발생시 멈추지 않고 해당 영역만 skip 한다(별도 처리 하지 않는 이상 정상 응답을 return한다).
• onErrorMap : 예외 발생시 다른 Exception으로 변환한다.

doOnError

Source:

@RequestMapping("/error/test")
public Flux<Integer> getNum() {
  return Flux.range(0, 5)
      .map(x -> {
        if (x == 3) {
          throw new IllegalArgumentException("invalid num");
        }
        return x;
      })
      .doOnError(x -> log.error("num generate error"))
      .log();
}

Result:

 INFO 5276 --- [ctor-http-nio-2] reactor.Flux.PeekFuseable.1              : | onSubscribe([Fuseable] FluxPeekFuseable.PeekFuseableSubscriber)
 INFO 5276 --- [ctor-http-nio-2] reactor.Flux.PeekFuseable.1              : | request(unbounded)
 INFO 5276 --- [ctor-http-nio-2] reactor.Flux.PeekFuseable.1              : | onNext(0)
 INFO 5276 --- [ctor-http-nio-2] reactor.Flux.PeekFuseable.1              : | onNext(1)
 INFO 5276 --- [ctor-http-nio-2] reactor.Flux.PeekFuseable.1              : | onNext(2)
ERROR 5276 --- [ctor-http-nio-2] c.e.demo.r4.controller.R4Controller      : num generate error
ERROR 5276 --- [ctor-http-nio-2] reactor.Flux.PeekFuseable.1              : | onError(java.lang.IllegalArgumentException: invalid num)
ERROR 5276 --- [ctor-http-nio-2] reactor.Flux.PeekFuseable.1              : 

onErrorReturn

Source:

@RequestMapping("/error/test")
public Flux<Integer> getNum() {
  return Flux.range(0, 5)
      .map(x -> {
        if (x == 3) {
          throw new IllegalArgumentException("invalid num");
        }
        return x;
      })
      .onErrorReturn(99)
      .log();
}

Result:

 INFO 12068 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onSubscribe(FluxOnErrorResume.ResumeSubscriber)
 INFO 12068 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : request(unbounded)
 INFO 12068 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onNext(0)
 INFO 12068 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onNext(1)
 INFO 12068 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onNext(2)
 INFO 12068 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onNext(99)
 INFO 12068 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onComplete()

onErrorResume

Source:

@RequestMapping("/error/test")
public Flux<Integer> getNum() {
  return Flux.range(0, 5)
      .map(x -> {
        if (x == 3) {
          throw new IllegalArgumentException("invalid num");
        }
        return x;
      })
      .onErrorResume(exec -> {
        log.error("num generate error");
        return Mono.just(99); //어떤 값으로 return을 할지 상황에 맞게 지정한다.
      })
      .log();
}

Result:

 INFO 8444 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onSubscribe(FluxOnErrorResume.ResumeSubscriber)
 INFO 8444 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : request(unbounded)
 INFO 8444 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onNext(0)
 INFO 8444 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onNext(1)
 INFO 8444 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onNext(2)
ERROR 8444 --- [ctor-http-nio-2] c.e.demo.r4.controller.R4Controller      : num generate error
 INFO 8444 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onNext(99)
 INFO 8444 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onComplete()

OnErrorContinue

Source:

@RequestMapping("/error/test")
public Flux<Integer> getNum() {
  return Flux.range(0, 5)
      .map(x -> {
        if (x == 3) {
          throw new IllegalArgumentException("invalid num");
        }
        return x;
      })
      .onErrorContinue((exec, num) ->
          log.error("num generate error, number is : {}, error msg : {}", num, exec.getMessage()))
      .log();
}

Result:

 INFO 9728 --- [ctor-http-nio-2] reactor.Flux.ContextStart.1              : | onSubscribe([Fuseable] FluxContextStart.ContextStartSubscriber)
 INFO 9728 --- [ctor-http-nio-2] reactor.Flux.ContextStart.1              : | request(unbounded)
 INFO 9728 --- [ctor-http-nio-2] reactor.Flux.ContextStart.1              : | onNext(0)
 INFO 9728 --- [ctor-http-nio-2] reactor.Flux.ContextStart.1              : | onNext(1)
 INFO 9728 --- [ctor-http-nio-2] reactor.Flux.ContextStart.1              : | onNext(2)
ERROR 9728 --- [ctor-http-nio-2] c.e.demo.r4.controller.R4Controller      : num generate error, number is : 3, error msg : invalid num
 INFO 9728 --- [ctor-http-nio-2] reactor.Flux.ContextStart.1              : | onNext(4)
 INFO 9728 --- [ctor-http-nio-2] reactor.Flux.ContextStart.1              : | onComplete()

onErrorMap

Source:

@RequestMapping("/error/test")
public Flux<Integer> getNum() {
  return Flux.range(0, 5)
      .map(x -> {
        if (x == 3) {
          throw new IllegalArgumentException("invalid num");
        }
        return x;
      })
      .onErrorMap(exec -> {
        log.error("num generate error, error msg : {}", exec.getMessage());
        throw new RuntimeException("change error type");
      })
      .log();
}

Result:

 INFO 14204 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onSubscribe(FluxOnErrorResume.ResumeSubscriber)
 INFO 14204 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : request(unbounded)
 INFO 14204 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onNext(0)
 INFO 14204 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onNext(1)
 INFO 14204 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onNext(2)
ERROR 14204 --- [ctor-http-nio-2] c.e.demo.r4.controller.R4Controller      : num generate error, error msg : invalid num
ERROR 14204 --- [ctor-http-nio-2] reactor.Flux.OnErrorResume.1             : onError(java.lang.RuntimeException: change error type)

[참고링크]
[Webflux Tip] 3. Webflux Error 처리! . (n.d.). https://akageun.github.io/2019/07/26/spring-webflux-tip-3.html.

Read More

MongoDB Array Query(With MongoTemplate)

Mongo Collection 구조

컬렉션명 : bookstore

{
  "_id": "1234567890",
  "books": [
    {
      "bookId": "1",
      "storeList": ["용산", "구로", "신도림A", "신도림B"]
    },
    {
      "bookId": "2",
      "storeList": ["서울역", "영등포A"]
    },
    {
      "bookId": "3",
      "bookNm": "재건축",
      "storeList": ["강남", "개포동"]
    }    
  ]
}

1. books element 추가

Monogo Query:

db.getCollection('bookstore').update(
  { _id: "1234567890" },
  { $push: { books: { bookId: "4", storeList: [ "신사", "대학로", "신촌" ] } } }
)

MongoTemplate:

var query = Query.query(Criteria.where("_id").is("1234567890"));
var update = new Update();
var book = new Book();
book.setBookId("4")
book.setStoreList(Arrays.asList("신사", "대학로", "신촌"));
update.push("books", book);

mongoTemplate.updateFirst(query, update, "bookstore");

2. books element 삭제

Monogo Query:

db.getCollection('bookstore').update(
  { _id: "1234567890" },
  { $pull: { books: { bookId: "4" } } }
)

MongoTemplate:

var query = Query.query(Criteria.where("_id").is("1234567890"));
var pullQuery = new Query();
pullQuery.addCriteria(Criteria.where("bookId").is("4"));
var update = new Update();
update.pull("books", pullQuery);

mongoTemplate.updateFirst(query, update, "bookstore");

3. storeList의 element 추가 - 값이 없는 경우에만

Monogo Query:

db.getCollection('bookstore').update(
  { _id: "1234567890", books: { $elemMatch: { bookId: "3" } } },
  { $addToSet: { "books.$.storeList": "신사동" } }
)

MongoTemplate:

var query = new Query();
var criteria = Criteria.where("_id").is("1234567890")
                  .and("books").elemMatch(Criteria.where("bookId").is("3"));
query.addCriteria(criteria);
var update = new Update();
update.addToSet("books.$.storeList", "신사동");

mongoTemplate.updateFirst(query, update, "bookstore");

4. storeList의 element 삭제

Monogo Query:

db.getCollection('bookstore').update(
  { _id: "1234567890", books: { $elemMatch: { bookId: "3" } } },
  { $pull: { "books.$.storeList": "신사동" } }
)

MongoTemplate:

var query = new Query();
var criteria = Criteria.where("_id").is("1234567890")
                  .and("books").elemMatch(Criteria.where("bookId").is("3"));
query.addCriteria(criteria);
var update = new Update();
update.pull("books.$.storeList", "신사동");

mongoTemplate.updateFirst(query, update, "bookstore");

5. storeList의 element 변경

Monogo Query:

db.getCollection('bookstore').updateOne(
    {_id: "1234567890", books: {$elemMatch:{bookId: "3"}}},
    {$set: {"books.$.bookNm": "재개발"}}
)

MongoTemplate:

var query = new Query();
var criteria = Criteria.where("_id").is("1234567890")
                  .and("books").elemMatch(Criteria.where("bookId").is("3"));
query.addCriteria(criteria);
var update = new Update();
update.set("books.$.bookNm", "재개발");

mongoTemplate.updateFirst(query, update, "bookstore");

Read More

Kafka session.timeout.ms와 max.poll.interval.ms의 차이

session.timeout.ms

컨슈머가 GroupCoordinator에게 해당시간내에 하트비트를 전송하지 않으면 리밸런싱이 일어난다. 컨슈머가 죽었으나 연결이 정상적으로 끊어지지 않는 상황이 발생하면 해당 값에 의해서 인지가 가능하다.
group.min.session.timeout.ms와 group.max.session.timeout.ms 범위 값만 설정가능
default : 10000(10초)

max.poll.interval.ms

다음 poll까지 대기하는 시간. 값이 300000ms인데 poll 내부로직이 300000ms가 넘어가는 경우 리밸런싱이 일어난다.
default : 300000(5분)

아래 코드 참고

@KafkaListener(topics = "xx", groupId = "yy")
public void kafkaListen(String message) throws InterruptedException {
    for (int x = 0; x < 10_000; x++) {
        Thread.sleep(15_000);
    }
}

컨슈머가 GroupCoordinator에게 하트비트는 전송하지만 실제로 아무일도 하지 않는 경우 컨슈머에 문제가 있는 것으로 판단하고 해당 파티션을 제외처리한다.

• 카프카 0.10.1 버전부터 폴링 간에도 하트비트를 전송할 수 있도록 별도의 하트비트 스레드가 추가되었다고 알고 있음
• 그래서 poll 내부로직이 오래 걸리더라도 하트비트는 계속 보내기 때문에 리밸런싱이 당연히 발생하지 않을 것으로 예상
• 하지만 poll 내부로직이 오래 걸리자 리밸런싱이 발생함
• 의아하게 생각하고 session.timeout.ms를 늘려봤지만 동일한 문제가 발생함
• 원인을 찾다보니 max.poll.interval.ms 시간을 초과했기 때문이였음
• poll 내부로직이 오래 걸리면 max.poll.interval.ms 시간을 늘리자

Read More