JacksonJsonDeserializer

Spring Kafka가 제공하는 JSON 역직렬화기로, Kafka 메시지의 value를 Jackson으로 Java/Kotlin 객체로 변환한다.

val javaType = jacksonObjectMapper().constructType(typeRef)
val deserializer = JacksonJsonDeserializer<T>(javaType).apply {
    addTrustedPackages("com.foo")
    setUseTypeHeaders(false)
}

addTrustedPackages

역직렬화 시 허용할 패키지를 지정하는 보안 설정이다.

• Spring Kafka의 JsonDeserializer는 기본적으로 임의의 클래스 역직렬화를 차단한다 (원격 코드 실행 등 보안 위협 방지).
• 역직렬화 대상 클래스가 신뢰된 패키지에 속하지 않으면 예외가 발생한다.
• addTrustedPackages("*")로 모든 패키지를 허용할 수도 있지만, 프로덕션에서는 필요한 패키지만 명시하는 것이 권장된다.

// com.foo 하위 클래스만 역직렬화 허용
addTrustedPackages("com.foo")

setUseTypeHeaders

Kafka 메시지 헤더에 포함된 타입 정보를 사용할지 여부를 설정한다.

• true (기본값): 메시지 헤더(__TypeId__)에 담긴 클래스 정보를 읽어 해당 타입으로 역직렬화한다. Producer가 JsonSerializer로 보낸 메시지에는 이 헤더가 자동으로 붙는다.
• false: 헤더를 무시하고, 생성자에서 지정한 타입으로만 역직렬화한다.

// 헤더 무시 → 항상 javaType으로 역직렬화
setUseTypeHeaders(false)

false로 설정하는 주요 사유는 다음과 같다.

• Producer가 다른 시스템이라 __TypeId__ 헤더가 없거나 다른 클래스명을 사용하는 경우
• 헤더의 타입 정보를 신뢰하지 않고 명시적으로 타입을 고정하고 싶은 경우

setRemoveTypeHeaders

역직렬화 후 타입 관련 헤더(__TypeId__, __KeyTypeId__, __ContentTypeId__)를 메시지에서 제거할지 여부를 설정한다.

• true (기본값): 역직렬화 후 타입 헤더를 제거한다.
• false: 타입 헤더를 그대로 유지한다. 하나의 메시지를 여러 Consumer가 순차적으로 읽거나, 다운스트림에서도 타입 정보가 필요한 경우에 사용한다.

setRemoveTypeHeaders(false)

setTypeMapper

기본 타입 매퍼(DefaultJackson2JavaTypeMapper)를 커스텀 매퍼로 교체한다. 타입 헤더의 해석 방식을 완전히 제어하고 싶을 때 사용한다.

val mapper = DefaultJackson2JavaTypeMapper().apply {
    setTypePrecedence(Jackson2JavaTypeMapper.TypePrecedence.TYPE_ID)
    addTrustedPackages("com.foo")
}
setTypeMapper(mapper)

setTypeFunction

바이트 배열과 헤더를 받아 동적으로 역직렬화 대상 JavaType을 결정하는 함수를 지정한다. 하나의 토픽에 여러 타입의 메시지가 섞여 있을 때 유용하다.

setTypeFunction { data, headers ->
    val typeHeader = String(headers.lastHeader("messageType").value())
    when (typeHeader) {
        "ORDER" -> jacksonObjectMapper().constructType(Order::class.java)
        "PAYMENT" -> jacksonObjectMapper().constructType(Payment::class.java)
        else -> jacksonObjectMapper().constructType(Map::class.java)
    }
}

setUseTypeMapperForKey

기본 타입 매퍼가 key 타입 헤더(__KeyTypeId__)를 참조하도록 설정한다. key를 JSON으로 역직렬화할 때 사용한다.

setUseTypeMapperForKey(true)

configure() 프로퍼티

setter 대신 Map<String, ?> 또는 Spring Boot 프로퍼티로 설정할 수도 있다.

# application.yml 예시
spring:
  kafka:
    consumer:
      properties:
        spring.json.trusted.packages: "com.foo"
        spring.json.use.type.headers: false
        spring.json.value.default.type: "com.foo.MyEvent"

주의: setter를 하나라도 호출한 뒤 configure()를 호출하면 프로퍼티 설정이 무시된다. setter 방식과 프로퍼티 방식을 혼용하지 않도록 한다.

Fluent API

메서드 체이닝을 지원하는 Fluent 스타일 메서드도 제공된다.

val deserializer = JacksonJsonDeserializer<MyEvent>(javaType)
    .trustedPackages("com.foo")
    .ignoreTypeHeaders()          // setUseTypeHeaders(false) 와 동일
    .dontRemoveTypeHeaders()      // setRemoveTypeHeaders(false) 와 동일
    .forKeys()                    // key 역직렬화용으로 지정
    .typeFunction { data, headers -> /* JavaType 반환 */ }

조합 예시

javaType을 생성자에서 직접 넘기고 헤더를 끄면, 항상 호출자가 지정한 타입으로 역직렬화된다.

val javaType = jacksonObjectMapper().constructType(typeRef)
val deserializer = JacksonJsonDeserializer<T>(javaType).apply {
    addTrustedPackages("com.foo")   // 이 패키지의 클래스만 역직렬화 허용
    setUseTypeHeaders(false)        // 헤더 타입 무시, 지정한 javaType으로 고정
}

타입 소거 (Type Erasure)

JVM에서 제네릭 타입 정보는 컴파일 이후 런타임에 소거된다. 따라서 아래와 같이 제네릭 타입을 직접 참조하는 코드는 컴파일 에러가 발생한다.

fun <T> printType() {
    println(T::class) // 컴파일 에러 - T의 타입을 알 수 없음
}

reified 키워드

inline 함수와 함께 reified를 사용하면 런타임에도 타입 정보를 유지할 수 있다.

inline fun <reified T> printType() {
    println(T::class) // 정상 동작
}

printType<String>() // class kotlin.String

왜 inline이 필요한가?

inline 함수는 호출 지점에 함수 본문이 복사된다. 컴파일러가 호출 시점에 실제 타입을 알고 있으므로, 복사할 때 T를 구체적인 타입으로 치환할 수 있다.

// 컴파일 전
printType<String>()

// 인라인 후 (개념적)
println(String::class)

활용 예시

타입 체크 / 캐스팅

inline fun <reified T> List<*>.filterByType(): List<T> {
    return this.filter { it is T }.map { it as T }
}

val mixed = listOf(1, "hello", 2, "world")
val strings = mixed.filterByType<String>() // ["hello", "world"]

Jackson/Gson 등 역직렬화

inline fun <reified T> String.fromJson(): T {
    return objectMapper.readValue(this, T::class.java)
}

val user = jsonString.fromJson<User>()

Android에서 클래스 참조 전달 생략

// reified 없이
startActivity(Intent(this, MainActivity::class.java))

// reified 사용
inline fun <reified T : Activity> Context.startActivity() {
    startActivity(Intent(this, T::class.java))
}
startActivity<MainActivity>()

제약사항

• 반드시 inline 함수에서만 사용 가능
• reified 타입 파라미터로 새 인스턴스 생성 불가 (T() 불가)
• Java에서 호출 불가 (인라인 메커니즘이 Kotlin 전용)

?. - 안전한 호출 (Safe calls)

• null이 아닌 경우 실행한다.
• let과 연계 사용시 더 간결하게 처리 가능

    // JAVA
    public String safetyUpperCase(String str) {
        if (str != null) {
            return str.toUpperCase();
        }
    }

    // kotlin
    fun safetyUpperCase(str:String?) = str?.uppercase()
    // str이 null일 경우 let 이하 구문은 실행되지 않는다
    fun combinePrefixIfNotNull(str:String?, prefix:String) = str?.let { prefix + str }

    val s1:String? = null
    val s2:String? = "A"
    combinePrefixIfNotNull(s1, "PREFIX-") // null
    combinePrefixIfNotNull(s2, "PREFIX-") // PREFIX-A

?: - 엘비스연산자 (Elvis operator)

• null 대신 사용 할 디폴트 값을 지정

    // JAVA
    String value = str == null ? "DEFAULT" : str;

    // kotlin
    val value = str?: "DEFAULT"

as? - 안전한 캐스트 (Safe casts)

• 지정한 타입으로 캐스트 하고 불가 시 null 반환

    // JAVA
    public Long toLong(Object obj) {
        if (obj instanceof Long) {
            return (Long)obj
        } else {
            return null;
        }
    }

    // kotlin
    fun toLong(obj:Any):Long? = obj as? Long

!! - 널 아님 단언 (not-null assertion)

• null이 될 수 있는 타입을 null이 될 수 없는 타입으로 강제전환
• 실제 값이 null일 경우 NPE 발생
• 안티패턴 : person.company!!.address!!.country ** 어느 값에서 NPE가 발생했는지 알기 어렵다

    fun <T> notNullAssert(nullable:T?):T = nullable!!

    val s1:String? = "NOT-NULL"
    val s2:String? = null
    notNullAssert(s1)
    notNullAssert(s2)  // NullPointerException

배치 구성을 위해 작업중 발생한 이슈로, Job이 모두 끝났음에도 불구하고 스프링 프로세스가 계속 유지되다 1분후에 종료되는 현상이 있었습니다.

구성

• Spring boot 2.3.2
• Spring batch
• Spring data jpa
• 기타 등등…

위와 같은 구성으로 배치 프로젝트를 구성하고 간단한 테스트 Job을 만들었습니다.

@EnableBatchProcessing
@SpringBootApplication
public class BatchApplication {
    public static void main(String[] args) {
        SpringApplication.run(BatchApplication.class, args);
    }
}

@Slf4j
@Configuration
@RequiredArgsConstructor
public class TestJobConfig {
    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    @JobScope
    public Step testStep() {
        return stepBuilderFactory.get("testStep")
            .tasklet((contribution, chunkContext) -> {

                ......

                return RepeatStatus.FINISHED;
            })
            .build();
    }

    @Bean
    public Job testJob() {
        return jobBuilderFactory.get("testJob")
            .start(testStep())
            .listener(listener())
            .build();
    }

    @Bean
    public JobExecutionListener listener() {
        return new JobExecutionListener() {
            @Override
            public void beforeJob(JobExecution jobExecution) {
                log.warn("Job Start !!!");
            }

            @Override
            public void afterJob(JobExecution jobExecution) {
                log.warn("Job End !!!");
            }
        };
    }
}

// build.gradle
dependencies {
    implementation "org.springframework.boot:spring-boot-starter-jdbc:${springBootVersion}"
    implementation "org.springframework.boot:spring-boot-starter-data-jpa:${springBootVersion}"
    implementation "org.springframework.boot:spring-boot-starter-batch:${springBootVersion}"

    ......

    testImplementation('org.springframework.boot:spring-boot-starter-test') {
        exclude group: 'org.junit.vintage', module: 'junit-vintage-engine'
    }

    testImplementation "org.springframework.batch:spring-batch-test"
}

테스트를 위해 별다른 로직 없이 간단하게 구성하였으며 시작과 종료시점에 로그를 남기 상태입니다. 현 상태에서 실행하게 되면 아래 로그와 같이 Job이 종료한 시점으로부터 1분이 지나서야 스프링 프로세스가 종료되는 것을 확인 할 수 있습니다.

2020-08-12 20:49:16 [restartedMain] INFO  o.s.b.a.b.JobLauncherApplicationRunner - Running default command line with: [requestDate=32]
2020-08-12 20:49:17 [restartedMain] INFO  o.s.b.c.l.support.SimpleJobLauncher - Job: [SimpleJob: [name=testJob]] launched with the following parameters: [{requestDate=32}]

// Job 시작 시
2020-08-12 20:49:17 [restartedMain] WARN  c.k.m.s.batch.config.TestJobConfig - Job Start !!!
2020-08-12 20:49:17 [restartedMain] INFO  o.s.batch.core.job.SimpleStepHandler - Executing step: [testStep]
2020-08-12 20:49:17 [restartedMain] WARN  c.k.m.s.batch.config.TestJobConfig - requestDate=>32
2020-08-12 20:49:17 [restartedMain] INFO  o.s.batch.core.step.AbstractStep - Step: [testStep] executed in 34ms
2020-08-12 20:49:17 [restartedMain] WARN  c.k.m.s.batch.config.TestJobConfig - Job End !!!
2020-08-12 20:49:17 [restartedMain] INFO  o.s.b.c.l.support.SimpleJobLauncher - Job: [SimpleJob: [name=testJob]] completed with the following parameters: [{requestDate=32}] and the following status: [COMPLETED] in 100ms
// Job 종료

// 스프링 프로세스 종료
2020-08-12 20:50:16 [SpringContextShutdownHook] INFO  o.s.o.j.LocalContainerEntityManagerFactoryBean - Closing JPA EntityManagerFactory for persistence unit 'default'
2020-08-12 20:50:16 [SpringContextShutdownHook] INFO  o.s.s.c.ThreadPoolTaskExecutor - Shutting down ExecutorService 'applicationTaskExecutor'
2020-08-12 20:50:16 [SpringContextShutdownHook] INFO  com.zaxxer.hikari.HikariDataSource - HikariPool-1 - Shutdown initiated...
2020-08-12 20:50:16 [SpringContextShutdownHook] INFO  com.zaxxer.hikari.HikariDataSource - HikariPool-1 - Shutdown completed.
Disconnected from the target VM, address: '127.0.0.1:60141', transport: 'socket'

Process finished with exit code 1

처음에는 Spring batch 쪽 이슈로 생각되었으나 공식 Github Issue에 올라온 내용에 따르면 Spring boot(jpa) 쪽 이슈라고 합니다.

원인은 Spring boot의 빠른 초기 기동을 위해 적용된 jpa의 bootstrap-mode 이슈로 application.yml에 다음과 같이 설정하여 해결 할 수 있습니다.

data.jpa.repositories.bootstrap-mode: default

위 기능은 초기화에 많은 시간을 잡아먹는 repository들의 초기화를 늦춰 boot의 기동시간을 단축 시키는 모드인데, 2.3.2 기준으로 기본값은 deferred로 지연로딩되고 백그라운드 스레드에 의해 특정 시점에 초기화 됩니다. 더 자세한 설명은 Spring Data JPA - Reference Documentation - Bootstrap Mode를 참고하세요.

(직접 디버깅을 하진 못했으나)가설을 세워보면 스프링 프로세스가 기동되면서 백그라운드 스레드는 특정 이벤트가 발생하는 시점까지 대기 하고 있는데, 배치의 Job이 먼저 끝나버리고 초기화가 끝나지 않은 스프링은 초기화를 마친 후 프로세스를 다시 종료한 것으로 유추되어 Job 실행을 1분이상 걸리도록 수정한 후 다시 테스트 하였습니다. 테스트 결과, 아래와 같이 Job이 종료된 이후 스프링 프로세스도 바로 종료 됨을 확인 할 수 있었습니다.

2020-08-12 21:26:37 [restartedMain] INFO  o.s.b.c.l.support.SimpleJobLauncher - Job: [SimpleJob: [name=testJob]] launched with the following parameters: [{requestDate=31}]

// Job 시작
2020-08-12 21:26:37 [restartedMain] WARN  c.k.m.s.batch.config.TestJobConfig - Job Start !!!
2020-08-12 21:26:37 [restartedMain] INFO  o.s.batch.core.job.SimpleStepHandler - Executing step: [testStep]
2020-08-12 21:26:37 [restartedMain] WARN  c.k.m.s.batch.config.TestJobConfig - requestDate=>31

// 2분여간 대기후 종료
2020-08-12 21:28:37 [restartedMain] INFO  o.s.batch.core.step.AbstractStep - Step: [testStep] executed in 2m0s53ms
2020-08-12 21:28:37 [restartedMain] WARN  c.k.m.s.batch.config.TestJobConfig - Job End !!!
2020-08-12 21:28:37 [restartedMain] INFO  o.s.b.c.l.support.SimpleJobLauncher - Job: [SimpleJob: [name=testJob]] completed with the following parameters: [{requestDate=31}] and the following status: [COMPLETED] in 2m0s121ms

// 스프링 프로세스 종료
2020-08-12 21:28:37 [SpringContextShutdownHook] INFO  o.s.o.j.LocalContainerEntityManagerFactoryBean - Closing JPA EntityManagerFactory for persistence unit 'default'
2020-08-12 21:28:37 [SpringContextShutdownHook] INFO  o.s.s.c.ThreadPoolTaskExecutor - Shutting down ExecutorService 'applicationTaskExecutor'
2020-08-12 21:28:37 [SpringContextShutdownHook] INFO  com.zaxxer.hikari.HikariDataSource - HikariPool-1 - Shutdown initiated...
2020-08-12 21:28:37 [SpringContextShutdownHook] INFO  com.zaxxer.hikari.HikariDataSource - HikariPool-1 - Shutdown completed.
Disconnected from the target VM, address: '127.0.0.1:60519', transport: 'socket'

Process finished with exit code 0

참고

Spring Batch - Spring Boot batch job does not shut down automatically after completion when using JPA Spring Boot - Spring Boot batch job does not shut down automatically after completion when using JPA

Triggering a ConcurrentModificationException

병렬 프로그래밍을 하다 보면 만나게 되는 흔한 오류 중 하나가 java.util.ConcurrentModificationException입니다. thread-safe 하지 않은 ArrayList 같은 객체를 여러 스레드에서 동시에 조작하다 보면 발생하게 되는데 주된 원인은 for-loop 도중 조작할 경우 입니다.

// java
List<Integer> intList = new ArrayList<>();
for (Integer i : intList) {
    ...
}

// class
List<Integer> intList = new ArrayList<>();
Iterator<Integer> iter = intList.iterator();
while(iter.hasNext()) {
    Integer i = iter.next();
    ...
}

for-loop 구문을 class 컴파일 하면 위와 같은 코드로 변환되는데 ArrayList가 Iterator를 생성할 때 내부 클래스인 ArrayList.Itr 클래스로 생성하여 반환되며 Itr 클래스는 생성시 ArrayList의 상위클래스인 AbstractList의 modCount 변수를 자신의 지역변수인 expectedModCount에 할당합니다. modCount는 ArrayList의 변경사항이 생길 때 마다 변경된 카운트를 기록하는 변수로 add / remove / trimToSize 등등 다수의 메서드에서 증가됩니다.

private class Itr implements Iterator<E> {
        int cursor;       // index of next element to return
        int lastRet = -1; // index of last element returned; -1 if no such
        int expectedModCount = modCount;
        ...
}

expectedModCount 변수는 next() 메서드를 실행할 때 기존 자신이 참조하고 있는 ArrayList의 변경사항이 있는지 체크하게 되는데 이 때 참조 리스트의 modCount와 자신의 지역변수인 expectedModCount를 비교하여 다를 경우 java.util.ConcurrentModificationException를 발생시킵니다.

// ArrayList.Itr class
public E next() {
    checkForComodification();
    ...
}

final void checkForComodification() {
    if (modCount != expectedModCount)
        throw new ConcurrentModificationException();
}

static 변수로 선언되어 여러 스레드에 조작되다 발생 할 수도 있으나 아래 예제와 같이 단일 스레드 내에서도 동일하게 발생 할 수 있습니다. 다른 블로그나 StackOverFlow의 질문들을 보면 대부분 remove 구문을 예시로 들지만 remove 뿐만 아닌 modCount가 조작되는 모든 메서드에 해당됩니다.

@Test(expected = ConcurrentModificationException.class)
public void concurrentModificationExceptionTest() {
    List<Integer> list = Lists.newArrayList(1,2,3,4,5,6,7);
    for (Integer i : list) {
        if (i==2) {
            list.add(8);
        }
    }
}

회피 하는 방법은 여러가지가 존재하는데 Avoiding the ConcurrentModificationException in Java문서를 참고하시면 좋습니다.

ConcurrentModificationException with Ehcache

자신이 병렬 스레드를 사용하지 않았고, for-loop문 내에서 조작하지 않았다고 해서 java.util.ConcurrentModificationException이 발생하지 않을꺼라 생각할 수 있으나 in-memory cache를 사용할 때 역시 주의해야 합니다.

// MyCacheService.class
@Cacheable
public List<Integer> getCacheList() { ... }

// OtherService
@Autowired private MyCacheService myCacheService;

public void sortList() {
    List<Integer> cachedList = myCacheService.getCacheList();
    cachedList.sort(Integer::compareTo);
    for(Integer i : cachedList) {
        System.out.println(i);
    }
}

얼핏 보기엔 sortList 메서드는 안전해 보이지만 Ehcache를 사용하고 있다면 부하 상황에서 반드시 java.util.ConcurrentModificationException이 발생하게 됩니다. for-loop 구문 안에서 조작하지 않았는데 발생하는 이유는 Ehcache가 데이터를 heap과 disk에 나눠 관리하기 때문입니다. 해당 데이터가 heap에 존재할 경우 캐시에서 동일한 레퍼런스 객체를 반환하기 때문에 static으로 공유된 객체를 사용하는 것과 동일한 이슈를 발생 시키게 되는데 아래 테스트케이스에서 볼 수 있듯이 캐시에서 받아온 List객체를 조작하게 되면 기존 캐시에 저장된 객체가 같이 바뀌어 있는것을 확인 할 수 있습니다.

@Test
public void cacheValueReferenceTest() {
    final Cache cache = getCache();
    final String key = "TEST_KEY";
    List<Integer> value = Lists.newArrayList(7,6,5,4,3,2,1);
    cache.put(key, value);
    
    value.sort(Integer::compareTo);
    
    List<Integer> cacheValue = cache.get(key, List.class);
    assertTrue(cacheValue.get(0) == 1);
}

동시성 문제가 발생하는 케이스는 아래 테스트케이스에서 확인 할 수 있습니다.

@Test
public void concurrentModifiedExceptionTest() throws InterruptedException {
    final Cache cache = getCache();
    final String key = "TEST_KEY";
    List<Integer> value = Lists.newArrayList(7,6,5,4,3,2,1);
    cache.put(key, value);

    AtomicBoolean isDetectException = new AtomicBoolean(false);
    CountDownLatch latch1 = new CountDownLatch(1);
    CountDownLatch latch2 = new CountDownLatch(1);
    
    Runnable thread1 = () -> {
        try {
            List<Integer> value1 = cache.get(key, List.class);
            for (Integer v : value1) {
                latch2.await();
            }
        } catch (Exception e) {
            log.error("thread1 error : {} {}", e.getClass().getName(), e.getMessage());
            isDetectException.set(e instanceof ConcurrentModificationException);
        } finally {
            latch1.countDown();
        }
    };
    
    Runnable thread2 = () -> {
        List<Integer> value2 = cache.get(key, List.class);
        value2.sort(Integer::compareTo);
        latch2.countDown();
    };

    new Thread(thread1).start();
    Thread.sleep(100);
    new Thread(thread2).start();
    latch1.await();
    assertTrue(isDetectException.get());
}

1번 스레드가 캐시에서 데이터를 가져와 for-loop 지점에 진입한 상태에서 2번 스레드가 캐시값에 접근하여 변경(sort와 같은..)작업을 시도하면 내부적으로 ArrayList의 modCount값이 바뀌면서 1번 스레드의 Iterator.next 구문 실행시 java.util.ConcurrentModificationException를 발생시키게 됩니다.

Conclusion

동시성 문제는 평시엔 잘 발견되지 않기 때문에 일반적인 테스트만 진행하고 실 서비스에 배포되면 운영중 가장 중요한 시기에 장애가 터지게 되어 난감할 때가 많습니다. 매번 성능 테스트를 할 수 없다면 위와 같은 경우에 항상 주의하며 코딩해야하며, 필자는 Ehcache의 프로세스를 확인하지 않아 디버깅에 오래걸렸지만 java.util.ConcurrentModificationException의 경우 반드시 for-loop 안에서 타켓 리스트객체를 수정했거나, 공유되는 리스트객체를 동시에 조작한 두가지 경우에 발생함을 유의하고 빠르게 디버깅 하시기 바랍니다.

Reference

Avoiding the ConcurrentModificationException in Java : https://www.baeldung.com/java-concurrentmodificationexception