Daehong Blog

◎ 개인 메모장
- Flink Table Maintenance
🏢 사내 이슈 해결
👨🏻‍💻 이력서
💻 개발 일지
📖 개인 공부
- 👉 Apache Hadoop
🖊️ 오픈소스 기여
- 👉 Apache Iceberg
  - 1. Iceberg Flink Catalog v2.0 MiniClusterWithClientResource 종속성 제거
  - 2. Flink 1.19 및 1.20에 MiniClusterWithClientResource 종속성 제거 Backport
- 👉 Spring Kafka
  - 1. Kafka RetryTopic 관련 기본 Template Bean 이름 변경
Home

Contact

Copyright © 2024 | Daehong

Home > 🖊️ 오픈소스 기여

Now Loading ...

🖊️ 오픈소스 기여

📘 Flink 1.19 및 1.20에 MiniClusterWithClientResource 종속성 제거 Backport
✏️ 1. 서론 이전 포스팅에서 소개한 것처럼, Iceberg Flink Catalog v2.0에서 MiniClusterWithClientResource 종속성 제거에 성공했습니다. 이에 따라 해당 이슈의 연장선이자 메인테이너의 요청에 따라, 동일한 변경 사항을 Flink 1.19와 1.20 버전에도 백포팅하기로 했습니다. (현재 Iceberg는 1.9.1 버전을 사용 중이며, Flink Catalog는 1.19, 1.20, 그리고 2.0 버전을 지원합니다.) 이번 작업은 이전 포스팅인 📘 Iceberg Flink Catalog v2.0 MiniClusterWithClientResource 종속성 제거 의 연장선이자, 유사한 내용이 많아, 비교적 짧게 정리해보려 합니다. ◈ ✏️ 2. 본론 ‍💻 2.1. PR: MiniClusterWithClientResource 종속성 제거 Backporting 이번 작업 역시 이전 PR의 연장선에 있기 때문에, 이슈 선정 과정에 대한 자세한 설명은 생략합니다. 아래와 같이, 백포팅에 대한 요청이 있었다는 내용을 남깁니다. 따라서 이번 포스팅에서는 곧바로 PR 과정부터 다루겠습니다. 이슈 선정 배경이나 PR 전체 흐름이 궁금하신 분들은 이전 포스팅인 📘 Iceberg Flink Catalog v2.0 MiniClusterWithClientResource 종속성 제거 을 참고해주시기 바랍니다. 1️⃣ COMMIT : v1.19 Backporting 기존 2.0의 TestIcebergSourceFailover 클래스와는 크게 다른 부분이 없었습니다. 중간에 구현된 코드가 약간은 다르긴 하였지만, 제가 수정해야하는 범위와는 무관한 부분이었기 때문에 넘어갔습니다. 해당 코드의 이전 버전은 아래와 같았습니다. ( v2.0 과 동일 ) Before: import org.apache.flink.test.util.MiniClusterWithClientResource; import org.apache.flink.util.function.ThrowingConsumer; // ... @Test public void testBoundedWithTaskManagerFailover() throws Exception { runTestWithNewMiniCluster( miniCluster -> testBoundedIcebergSource(FailoverType.TM, miniCluster)); } @Test public void testBoundedWithJobManagerFailover() throws Exception { runTestWithNewMiniCluster( miniCluster -> testBoundedIcebergSource(FailoverType.JM, miniCluster)); } // ... @Test public void testContinuousWithTaskManagerFailover() throws Exception { runTestWithNewMiniCluster( miniCluster -> testContinuousIcebergSource(FailoverType.TM, miniCluster)); } @Test public void testContinuousWithJobManagerFailover() throws Exception { runTestWithNewMiniCluster( miniCluster -> testContinuousIcebergSource(FailoverType.JM, miniCluster)); } // ... private static void runTestWithNewMiniCluster(ThrowingConsumer<MiniCluster, Exception> testMethod) throws Exception { MiniClusterWithClientResource miniCluster = null; try { miniCluster = new MiniClusterWithClientResource(MINI_CLUSTER_RESOURCE_CONFIG); miniCluster.before(); testMethod.accept(miniCluster.getMiniCluster()); } finally { if (miniCluster != null) { miniCluster.after(); } } } 그래서 아래와 같은 코드로 수정을 동일하게 진행하였습니다. After: import org.apache.flink.test.junit5.InjectMiniCluster; import org.junit.jupiter.api.AfterEach; // ... @BeforeEach protected void startMiniCluster(@InjectMiniCluster MiniCluster miniCluster) throws Exception { if (!miniCluster.isRunning()) { miniCluster.start(); } } @AfterEach protected void stopMiniCluster(@InjectMiniCluster MiniCluster miniCluster) throws Exception { miniCluster.close(); } // ... @Test public void testBoundedWithTaskManagerFailover(@InjectMiniCluster MiniCluster miniCluster) throws Exception { testBoundedIcebergSource(FailoverType.TM, miniCluster); } @Test public void testBoundedWithJobManagerFailover(@InjectMiniCluster MiniCluster miniCluster) throws Exception { testBoundedIcebergSource(FailoverType.JM, miniCluster); } // ... @Test public void testContinuousWithTaskManagerFailover(@InjectMiniCluster MiniCluster miniCluster) throws Exception { testContinuousIcebergSource(FailoverType.TM, miniCluster); } @Test public void testContinuousWithJobManagerFailover(@InjectMiniCluster MiniCluster miniCluster) throws Exception { testContinuousIcebergSource(FailoverType.JM, miniCluster); } 이렇게 코드를 수정하게 된 이유와, Minicluster 의 역할 그리고 InjectMiniCluster 와의 차이에 대해서 역시 이전 포스팅에 작성 되어있습니다. 거듭 말씀드리지만, 이전 포스팅을 꼭 참고하시기 바랍니다. :D 2️⃣ COMMIT : v1.20 Backporting v1.20 백포팅 작업은, v1.19 와 완전 동일하게 진행되었습니다. 코드 역시 완전 동일하게 수정하였습니다. 그리하여, 최종적으로는 아래와 같은 PR 을 전송하였고, 성공적으로 Merge 가 되었습니다. 👉 My PR: Backporting Removal of MiniClusterWithClientResource from Iceberg Flink Catalog v1.19, v1.20 🍹 중간에 마주한 약간의 이슈 이전 포스팅에서 Local 테스트와, CI 테스트를 성공하기 위하여 필요한 몇 가지 Gradle 명령어를 말씀드렸습니다. 이번에 역시 동일한 과정을 거쳤고, 코드 스타일을 맞추기 위해 ./gradlew :iceberg-flink:iceberg-flink-1.19:spotlessApply 와 ./gradlew :iceberg-flink:iceberg-flink-1.20:spotlessApply 명령어를 사용하였습니다. 그랬더니 아래와 같은 오류가 발생하게 됩니다. 에러를 보면, Gradle이 :iceberg-flink:iceberg-flink-1.20:spotlessApply 라는 태스크(혹은 모듈)를 실행하려 했지만, iceberg-flink 프로젝트 아래에 iceberg-flink-1.20 이라는 서브프로젝트가 존재하지 않다는 내용입니다. 존재하는 서브프로젝트 후보는 iceberg-flink-2.0 이 있다고 친절하게 알려주고 있지만, 우리에게 필요한건 1.19 와 1.20 버전입니다. 저는 처음에는 2.0 으로 해도, 자동으로 1.19 와 1.20 코드 스타일도 맞춰질 줄 알았는데 그게 아니었습니다. 그래서 아래와 같이 질문을 하였습니다. 그래서 ./gradlew properties 명령어를 통해 찾아보았더니, 역시 systemProp.defaultFlinkVersions: 2.0 으로, 되어있었습니다. 그래서 vi 로 gradle.properties 를 열어주었고, enableFlink1.19=true 과, enableFlink1.20=true 을 추가해준 뒤 spotlessApply 명령어를 실행해주었습니다. 그랬더니 정상적으로 성공하였고, PR 후 Merge 까지 성공하였습니다. ◈ ✏️ 3. 결론 이번 작업은 Iceberg Flink Catalog v2.0에서 진행했던 MiniClusterWithClientResource 제거 작업을 Flink 1.19 및 1.20 버전에 성공적으로 백포팅한 사례였습니다. 코드 변경 자체는 이전과 거의 동일했지만, Gradle 설정 및 로컬 테스트 환경 설정에서 발생한 작은 이슈를 해결하면서, Iceberg 프로젝트의 Gradle 구성과 서브모듈 활성화 방식에 대해 더 깊이 이해할 수 있는 계기가 되었습니다. 이번 경험을 바탕으로 앞으로도 오픈소스 프로젝트에 더욱 적극적으로 기여하며, 다양한 문제 상황을 주도적으로 해결해나갈 수 있도록 꾸준히 역량을 키워가겠습니다. 💪
🖊️ 오픈소스 기여 · 2025-05-27
📘 Iceberg Flink Catalog v2.0 MiniClusterWithClientResource 종속성 제거
✏️ 1. 서론 작년 10월경, 첫 오픈 소스 멘토링 프로그램에 참여하면서 Spring Kafka 프로젝트의 Contributor 가 되는 소중한 경험을 했습니다. 당시에는 오픈 소스 생태계가 처음이었기 때문에, 다소 낯설고 어렵게 느껴졌지만, 운 좋게도 입문자들을 위한 비교적 단순한 이슈를 통해 기여를 시작할 수 있었습니다. 그 이슈는 문서(Docs)의 정정과 일부 간단한 코드 수정 정도였지만, 저에게는 오픈 소스에 첫 발을 내딛는 데 매우 의미 있는 기회였습니다. 그리고 올해, 다시 한 번 같은 멘토링 시스템에 참여하게 되었습니다. 하지만 이번에는 멘토링을 받는 입장이 아니라, 멘토링을 지원하고 운영하는 입장으로 참여하게 되었습니다. 멘티들의 이슈 선정과 PR 과정에 조언을 주면서, 자연스럽게 저도 다시 한 번 오픈 소스 기여를 도전해보고 싶은 의욕이 생겼습니다. ( 이번 포스팅은 ICeberg 기여 포스팅이기 때문에, 멘토링에서 운영진으로서 활동한 내용은 최소화 하였습니다. 결론에만 살짝 포스팅하겠습니다. :D ) 그 과정에서 눈길이 간 프로젝트가 바로 Apache Iceberg였습니다. 이 프로젝트는 단순히 관심만 있었던 것이 아니라, 회사 내부에서 신규 솔루션 도입 시 직접 Iceberg를 제안하여 실제로 적용되기도 했고, 해당 기술을 바탕으로 사내 컨퍼런스 발표까지 진행했던 만큼, 개인적으로도 의미가 깊고 애착이 있는 프로젝트입니다. 그래서 이번에는 단순한 문서 수정에 머무르지 않고, 실질적인 코드 수준의 기여를 목표로 삼았습니다. Apache Iceberg의 구조를 더 깊이 있게 이해하고, 실제 동작을 분석하면서 프로젝트에 기여할 수 있는 포인트를 찾고자 했습니다. 이번 기회를 통해 더 큰 도전과 성장의 계기를 만들고 싶었습니다. ◈ ✏️ 2. 본론 🤔 2.1. 이슈 선택 과정 이번 이슈 선정 과정은 AI의 도움 50%, 그리고 저의 판단 50%로 이루어졌습니다. 먼저, 기여하고 싶은 프로젝트로는 단연 Apache Iceberg가 1순위였습니다. Iceberg 외에도 Spark, Airflow, Hadoop, Flink 등의 Apache 재단 프로젝트를 후보군에 두고, 기여 가능한 이슈들을 하나씩 탐색하기 시작했습니다. 이 과정에서 Google의 Gemini를 활용해 Iceberg의 GitHub 저장소에서 이슈 리스트를 수집했고, AI에게 “기여하기 좋은 이슈의 기준”을 프롬프트로 제시하여 필터링 작업을 진행했습니다. 그 결과, 여러 이슈 중에서 good first issue 라벨이 붙어 있는 항목들을 선별해낼 수 있었고, 그중에서도 단번에 눈에 띄는 이슈 하나를 발견하게 되었습니다. 이처럼 이번에는 수작업으로만 탐색했던 과거와 달리, AI 도구를 병행하여 시간을 절약하면서도 효율적으로 기여 대상 이슈를 선정할 수 있었습니다. 이렇게 이슈를 선택한 저는, 이슈를 올린 Maintainer 와의 소통을 통하여 해당 이슈를 맡겠다고 요청하였습니다. 이러한 과정을 통하여 Exclude JUnit4 dependency from classpath 에서 Remove JUnit4 dependency from Flink 까지 이어지는 해당 이슈에 기여를 하기로 확정하였고, 회사에서도 관심을 가지고, 저 또한 최근 공부를 하고 있던 Apache Iceberg 에 기여할 수 있는 기회를 잡게 되었습니다. 📚 2.2. 이슈: Iceberg Flink Catalog 의 Junit4 의존성 제거 Exclude JUnit4 dependency from classpath 와, Remove JUnit4 dependency from Flink 이슈 를 보면, Apache Iceberg의 Maintainer는 전체 코드베이스, 특히 Flink 관련 모듈에서 JUnit4 의존성을 완전히 제거하려는 명확한 의지를 드러내고 있었습니다. 두 이슈 모두 공통적으로, 테스트 코드가 JUnit5 기반으로 통일되길 원하며, 오래된 테스트 유틸이나 라이브러리에 묶여 있는 잔존 JUnit4 종속성을 제거하는 데 목적이 있었습니다. 먼저 #12937 이슈에서는 iceberg-flink 모듈 내에서 JUnit4 관련 종속성을 제거하고, JUnit5 스타일로 마이그레이션한 일부 작업 내역이 공유되어 있었습니다. 예를 들어 Assume.assumeFalse(…)와 같은 코드는 assumeThat(…).isEqualTo(…)와 같은 JUnit5 스타일로 대체되고 있었으며, build.gradle 파일에서는 junit 그룹을 명시적으로 exclude 처리하고 있었습니다. 하지만 아직도 완전히 제거하지 못한 부분들이 존재했습니다. 예를 들어, TestIcebergSourceFailover 테스트는 내부적으로 Flink의 MiniClusterWithClientResource를 사용하고 있었고, 이 유틸리티는 JUnit4에 의존하고 있기 때문에 완전한 제거에는 제한이 있었습니다. 또 다른 이슈인 #13049에서는 프로젝트 전체 build.gradle의 classpath 레벨에서 JUnit4를 명시적으로 제외하자는 제안이 담겨 있었습니다. 이 역시 Flink 모듈뿐만 아니라, Testcontainers 라이브러리의 일부 클래스(GenericContainer)도 여전히 JUnit4에 의존하고 있어, 완전한 제거를 위해서는 외부 라이브러리의 업데이트를 기다려야 하는 상황이었습니다. 이러한 배경에서, 저는 해당 이슈를 기반으로 구체적인 기여 방향을 정하고 작업을 시작하게 되었습니다. 단순한 문서 수정이나 설정 변경을 넘어서, 직접 테스트 코드의 구조를 개선하고, Gradle 의존성을 다루며, 라이브러리 호환성 문제까지 고민해야 하는 도전적인 과제였기에 더욱 의미가 컸습니다. 👉 해당 이슈는 위에서 설명한, Apache Iceberg 의 #13049 와, #12937 이슈 에 있는 내용입니다. 💬 2.3. Maintainer 와의 소통 과정을 통한, PR 방향성 확립 저는 Apache Iceberg 프로젝트의 Maintainer인 @nastra 님과 주요 기여자 중 한 분인 @tomtongue 님과 바로 소통을 시작했습니다. 먼저, 해당 작업에 참여하고 싶다는 의사를 댓글로 남겼고, Maintainer님께서는 매우 환영해 주시며 자유롭게 작업을 진행해도 된다는 긍정적인 답변을 주셨습니다. 또한, 저보다 먼저 같은 이슈에 관심을 보인 tomtongue 님과 협업 가능성도 열어두었는데, 그분은 Flink 마이그레이션 작업에는 시간이 좀 더 필요할 것 같다고 말씀해 주셨습니다. 이에 자연스럽게 제가 먼저 작업을 진행하는 방향으로 역할이 정리되었습니다. 구체적인 작업 계획으로는, 우선 JUnit4에 의존하고 있던 TestIcebergSourceFailover 클래스를 JUnit5로 마이그레이션하는 것과, Gradle 설정에서 JUnit4 종속성을 제거하는 작업을 포함했습니다. 처음에는 Flink 2.0 버전부터 작업을 시작한 뒤, 이후 1.20 및 1.19 버전으로 순차적으로 백포팅(backport)하겠다는 계획을 Maintainer님께 공유했습니다. Maintainer님도 이 계획이 적절하다며, 우선 Flink 2.0 모듈에서 작업을 진행한 뒤 결과를 검토받는 것이 좋겠다는 피드백을 주셨습니다. 이후 저는 첫 번째 PR인 #13016을 제출하였고, 기존에 사용되던 MiniClusterWithClientResource 의존성을 제거하면서 테스트 코드와 Gradle 설정을 함께 수정했습니다. 하지만 Gradle 설정을 수정하는 과정에서 예상치 못한 문제가 발생했습니다. 여러 가지 방법을 시도했음에도 불구하고, JUnit4에 대한 종속성을 완전히 제거하는 것이 쉽지 않았고, 빌드와 테스트 과정에서 지속적으로 오류가 발생했습니다. 이에 Maintainer님과 다시 소통한 결과, Flink 내부 유틸리티 중 하나인 InternalMiniClusterExtension이 여전히 JUnit4를 필요로 하기 때문에 완전한 제거는 현실적으로 어렵다는 점에 함께 동의하게 되었습니다. 그럼에도 불구하고, Iceberg 코드베이스 전체에서는 JUnit4 의존성을 최소화하고, 가능하면 JUnit5 중심으로 통일하는 방향이 바람직하다는 의견을 주셔서, 저 역시 이 가이드라인에 맞춰 TestIcebergSourceFailover 클래스를 중심으로 PR을 지속해서 정제해 나갈 수 있었습니다. 이와 같은 소통 과정을 통해, 단순한 코드 수정뿐만 아니라 프로젝트 내부 상황과 제약을 이해하며 협력하는 경험을 쌓을 수 있었습니다. ‍💻 2.4. PR: JUnit4 의존성 최소화 및 MiniClusterWithClientResource 종속성 제거 해당 소스를 수정하기 전, 저는 Iceberg에 대해서는 사내 컨퍼런스를 진행할 만큼 충분히 공부한 상태였지만, 카탈로그 엔진과 관련해서는 Spark와 JDBC 기반의 몇 가지에 대해서만 주로 알고 있었습니다. 반면, Flink에 대해서는 상대적으로 익숙하지 않았습니다. Flink는 주로 실시간 데이터 스트리밍 처리를 위해 사용되는 분산 처리 엔진으로, Iceberg에서는 Flink 커넥터를 통해 대용량 데이터를 효율적으로 처리하는 데 활용되고 있다 정도로만 알고 있었습니다. ( 저희 회사는 Flink 를 쓰고 있지는 않지만, Iceberg 를 공부할 때 참고하였던 Kakao 기술 블로그의 louis 님 글 덕분에 Flink 에 대하여 공부가 되었습니다. Apache Iceberg와 Flink CDC 심층 탐구, 아파치 플링크와 CDC의 만남. 플링크 CDC 맛보기 ) 하지만 Flink 내부에서 테스트를 위해 사용되는 ‘MiniCluster’라는 개념에 대해서는 전혀 알지 못했습니다. 이에 Flink 공식 문서와 커뮤니티 자료, 그리고 GitHub 저장소 등을 참고하여 ‘MiniCluster’가 무엇인지 공부하기 시작했습니다. MiniCluster는 Flink 클러스터를 로컬 환경에서 가볍게 실행할 수 있도록 해주는 일종의 미니 버전 클러스터로, 테스트 환경에서 실제 분산 환경과 유사한 조건을 만들어 Flink 작업을 검증할 수 있도록 돕는 역할을 합니다. 즉, 실제 클러스터를 띄우지 않고도 로컬에서 Flink 잡을 실행해볼 수 있게 해주기 때문에 테스트 자동화와 디버깅에 매우 유용합니다. 이처럼 MiniCluster가 Flink 테스트 환경에서 어떤 핵심적인 역할을 하는지 이해하는 것이, 이번 작업을 진행하는 데 중요한 첫걸음이 되었습니다. 그 다음은, 드디어 코드를 수정하기 시작하였습니다. 순서는 Junit4 를 어쩔 수 없이 의존해야하는 부분은 살리고, 이전 커미터가 미처 수정하지 못한 Junit5 로 바꿀 수 있는 부분을 수정하는 작업을 먼저 진행하였습니다. 이후에는, TestIcebergSourceFailover 클래스에서 Junit4 를 의존하고 있는 부분이 어디인지를 찾고, 그 부분을 Junit5 형태로 변환하는 작업을 진행하였습니다. 1️⃣ COMMIT : GenericAppenderHelper 의 생성자 @Deprecated 제거 첫 번째로, GenericAppenderHelper.java 코드에서, 생성자가 @Deprecated 되어있었습니다. 과거에 JUnit에서 흔히 쓰던 TemporaryFolder 객체를 인자로 받습니다. TemporaryFolder는 테스트 중 임시 파일/디렉토리를 자동 생성하고 정리해주는 도구입니다. 아마, Junit4 의존성을 완전히 걷어내기 위한 과정에서 @Deprecated 를 해둔 거 같습니다. 하지만 위에서 이야기하였듯 현재 Junit4 를 완전히 걷어내는 것은 어렵다고 판단하였기 때문에, 아직까지 GenericAppenderHelper 클래스를 필요로 하는 소스들을 위해서, @Deprecated 를 삭제해주었습니다. ( 삭제하지 않으면, CI 테스트 시 에러가 발생합니다. ) Before: @Deprecated public GenericAppenderHelper(Table table, FileFormat fileFormat, TemporaryFolder tmp, Configuration conf) { this.table = table; this.fileFormat = fileFormat; this.temp = tmp.getRoot().toPath(); this.conf = conf; } @Deprecated public GenericAppenderHelper(Table table, FileFormat fileFormat, TemporaryFolder tmp) { this(table, fileFormat, tmp, null); } After: public GenericAppenderHelper(Table table, FileFormat fileFormat, TemporaryFolder tmp, Configuration conf) { this.table = table; this.fileFormat = fileFormat; this.temp = tmp.getRoot().toPath(); this.conf = conf; } public GenericAppenderHelper(Table table, FileFormat fileFormat, TemporaryFolder tmp) { this(table, fileFormat, tmp, null); } 2️⃣ COMMIT : Junit4 의 Assume 을 걷어내고, AssertJ 방식 활용 두 번째로 이전에 Junit4 에서 Junit5 로 의존성 변경을 시도했던 커미터분이 미처 수정하지 못한 부분을 수정하는 과정을 거쳤습니다. Flink 2.0 의 TestIcebergSink 클래스에서, 아래 코드처럼 Junit4 방식의 Assume.assumeFalse(...) 을 사용하는 부분이 있어서, 다른 버전의 TestIcebergSink 클래스들과 동일하게 AssertJ 라이브러리의 assumeThat 방식으로 변경해주었습니다. Before: import org.junit.Assume; // ... @TestTemplate void testErrorOnNullForRequiredField() throws Exception { Assume.assumeFalse( "ORC file format supports null values even for required fields.", format == FileFormat.ORC); // Next Code ... } After: import static org.assertj.core.api.Assumptions.assumeThat; // ... @TestTemplate void testErrorOnNullForRequiredField() throws Exception { assumeThat(format) .as("ORC file format supports null values even for required fields.") .isNotEqualTo(FileFormat.ORC); // Next Code ... } 3️⃣ COMMIT : MiniClusterWithClientResource 종속성 제거 마지막이 드디어, 이번 PR 의 꽃인, TestIcebergSourceFailover 클래스에서 MiniClusterWithClientResource 종속성을 제거하는 것이었습니다. 이 부분에 대한 설명은 조금 길어질 수 있으니, 먼저 ASIS 코드를 보겠습니다. Before: import org.apache.flink.test.util.MiniClusterWithClientResource; import org.apache.flink.util.function.ThrowingConsumer; // ... @Test public void testBoundedWithTaskManagerFailover() throws Exception { runTestWithNewMiniCluster( miniCluster -> testBoundedIcebergSource(FailoverType.TM, miniCluster)); } @Test public void testBoundedWithJobManagerFailover() throws Exception { runTestWithNewMiniCluster( miniCluster -> testBoundedIcebergSource(FailoverType.JM, miniCluster)); } // ... @Test public void testContinuousWithTaskManagerFailover() throws Exception { runTestWithNewMiniCluster( miniCluster -> testContinuousIcebergSource(FailoverType.TM, miniCluster)); } @Test public void testContinuousWithJobManagerFailover() throws Exception { runTestWithNewMiniCluster( miniCluster -> testContinuousIcebergSource(FailoverType.JM, miniCluster)); } // ... private static void runTestWithNewMiniCluster(ThrowingConsumer<MiniCluster, Exception> testMethod) throws Exception { MiniClusterWithClientResource miniCluster = null; try { miniCluster = new MiniClusterWithClientResource(MINI_CLUSTER_RESOURCE_CONFIG); miniCluster.before(); testMethod.accept(miniCluster.getMiniCluster()); } finally { if (miniCluster != null) { miniCluster.after(); } } } 먼저, JUnit 5에서는 @BeforeAll, @AfterAll, @BeforeEach, @AfterEach를 사용해 리소스 생명 주기를 다루는 게 일반적입니다. 하지만 해당 코드는 테스트마다 miniCluster.before() / after()를 수동으로 호출합니다. 이부분이 JUnit 4 스타일이라고 할 수 있습니다. 또한, MiniClusterWithClientResource 는 보통 JUnit4의 ExternalResource 를 상속해서 구현합니다. JUnit4에서는 리소스 초기화와 정리를 위해 @Rule 어노테이션을 통해 ExternalResource 를 사용합니다. 이는, Flink 개발 초기 시점과 JUnit5 도입 시점의 시기적 차이 때문에, 기존에 잘 만들어진 테스트 유틸리티들이 JUnit4 스타일로 많이 남아 있습니다. 이런 이유 때문에, 사실 완전히 Junit4 를 당장 걷어내는 것이 어렵다고 판단한 거 같습니다. 즉, Flink 의 Minicluster 를 활용한 테스트코드 자체가 완전히 JUnit5로 완전히 마이그레이션되지 않은 상태이기 때문에, Iceberg 에도 당장 적용하기 어렵다는 것으로 생각됩니다. 계속 이어서 설명하자면, 여기서는 테스트가 실행될 때마다 MiniClusterWithClientResource 인스턴스를 직접 생성하고, 수동으로 before()와 after()를 호출해서 클러스터 시작과 종료를 명시적으로 제어하고 있습니다. 테스트 메서드들은 MiniCluster를 직접 파라미터로 받지 않고, 람다 ThrowingConsumer에 넘겨서 간접적으로 처리합니다. 즉, 리소스 관리를 테스트 코드 내부에서 직접 제어하는 형태입니다. 보통 이런 방식은 JUnit4에서 @Rule 같은 자동 리소스 관리가 없거나 커스텀 상황에서 쓰입니다. 그래서 아래와 같은 코드로 수정을 진행하였습니다. After: import org.apache.flink.test.junit5.InjectMiniCluster; // ... @Test public void testBoundedWithTaskManagerFailover(@InjectMiniCluster MiniCluster miniCluster) throws Exception { testBoundedIcebergSource(FailoverType.TM, miniCluster); } @Test public void testBoundedWithJobManagerFailover(@InjectMiniCluster MiniCluster miniCluster) throws Exception { testBoundedIcebergSource(FailoverType.JM, miniCluster); } // ... @Test public void testContinuousWithTaskManagerFailover(@InjectMiniCluster MiniCluster miniCluster) throws Exception { testContinuousIcebergSource(FailoverType.TM, miniCluster); } @Test public void testContinuousWithJobManagerFailover(@InjectMiniCluster MiniCluster miniCluster) throws Exception { testContinuousIcebergSource(FailoverType.JM, miniCluster); } 중점은, 리소스 관리면에서, MiniClusterWithClientResource 수동 생성 및 before()/after() 호출 하는 부분을 @InjectMiniCluster를 통해 자동으로 주입 및 관리를 할 수 있게끔 변경했다는 부분입니다. 그리고 테스트 코드 작성 시, 람다 방식으로 테스트 실행, 리소스 관리를 직접 제어하는 방식에서 테스트 메서드 파라미터로 리소스 주입 받아 사용하도록 변경하였고, ExternalResource 기반 (JUnit4 Rule) 방식에서 ParameterResolver 또는 Extension 기반 (JUnit5) 방식으로 변경하였다는 점이 있겠습니다. 이러한 방식에 MainTainer 들은 토론을 하였고, 괜찮은 방식이라는 결론에 다다르게 됩니다. 그리고, 다른 컨트리뷰터들 역시, 괜찮다는 리뷰를 달아주었습니다. 🔧 MiniCluster 주입 후 실행 오류 발생 및 수동 생명주기 관리 코드 추가 그러나, 이부분에 대하여 처음에는 CI 테스트에서 문제가 발생하였습니다. 테스트가 120초 안에 끝나지 않고 Timeout으로 실패하는 문제가 발생한 것입니다. @InjectMiniCluster를 통해 MiniCluster를 테스트 메서드에 주입받았지만 주입만 된 상태이지, 명시적으로 start()를 호출하지 않으면 MiniCluster가 실행되지 않았던 것으로 보입니다. 결국 테스트 로직에서 MiniCluster를 사용하려 했을 때, 클러스터가 “비활성 상태”이므로, 작업이 시작되지 않거나, 실행 중 hang 이 발생한 것으로 보입니다. 좀 더 원인을 찾아보니. JUnit5의 확장 모델에서는 @Inject… 같은 어노테이션으로 객체 주입은 가능하지만, 해당 객체의 생명주기(start/stop)는 자동으로 보장되지 않을 수 있다고 합니다. 또한 MiniCluster는 명시적으로 .start() 호출이 필요하고, 주입은 단순히 생성만 해주는 것이며, 실행은 별도 단계라는 원인을 찾을 수 있었습니다. 결과적으로 테스트가 MiniCluster에 작업을 제출하려 했지만, MiniCluster가 시작되지 않았고, 이는 테스트 timeout (120초) 초과로 실패까지 이어지게 된 것입니다. 그래서 TestIcebergSourceFailover 클래스에 아래와 같은 코드를 추가 커밋하였습니다. Add Code: import org.junit.jupiter.api.AfterEach; // AfterEach 추가 @BeforeEach protected void startMiniCluster(@InjectMiniCluster MiniCluster miniCluster) throws Exception { if (!miniCluster.isRunning()) { miniCluster.start(); } } @AfterEach protected void stopMiniCluster(@InjectMiniCluster MiniCluster miniCluster) throws Exception { miniCluster.close(); } @BeforeEach와 @AfterEach를 통해 MiniCluster의 시작과 종료를 명시적으로 제어함으로써, 테스트 실행 시 필요한 Flink 클러스터 환경이 정상적으로 초기화되고 정리될 수 있게 하였습니다. 그리고 그 결과, 최종적으로는 Iceberg 에 PR Merge 를 성공하여, Apache 재단의 Iceberg 프로젝트에 기여할 수 있었고, Contributor 가 될 수 있었습니다. 👉 My PR: Iceberg Flink Catalog v2.0 Remove the MiniClusterWithClientResource dependency 🍹 Local Test 및 CI 통과를 위한 절차 안내 처음 Iceberg 프로젝트에 기여하면서 가장 헷갈렸던 부분 중 하나가, 로컬 테스트와 CI 통과 방식이었습니다. 혹시 저처럼 Iceberg에 기여하고자 하는 분들이 계시다면, 아래 절차를 참고하시면 도움이 될 것입니다. ✅ 코드 스타일 정리: spotlessApply Iceberg는 Spotless를 사용하여 코드 포맷팅을 검사합니다. 따라서 PR을 생성하기 전 반드시 아래 명령어를 실행하여 코드 스타일을 자동으로 정리해야 합니다: ./gradlew :iceberg-flink:iceberg-flink-2.0:spotlessApply 이 명령어는 Java 코드의 정해진 스타일 가이드에 맞게 들여쓰기, 정렬, 공백 등을 자동으로 수정해줍니다. 이를 적용하지 않으면 CI에서 spotlessCheck 단계에서 실패하게 됩니다. ✅ 로컬 테스트 수행: test –stacktrace CI에 올리기 전에, 로컬 환경에서 먼저 유닛 테스트를 실행하여 이상이 없는지 확인해야 합니다: ./gradlew :iceberg-flink:iceberg-flink-2.0:test --stacktrace –stacktrace 옵션은 만약 테스트 실패 시 상세한 오류 메시지를 확인하는 데 도움이 됩니다. 테스트를 통과하지 못하면 CI 단계에서도 동일하게 실패하므로, 로컬에서 반드시 먼저 확인하는 것이 좋습니다. 참고로, 저는 flink 2.0 에 대하여 테스트를 진행하는 것이기 때문에 위와 같이 하였습니다. ./gradlew projects 를 통하여 테스트가 가능한 프로젝트 목록을 확인한 후에, 원하는 프로젝트를 가지고 진행해주면 되겠습니다. ◈ ✏️ 3. 결론 이번 기여를 통해 기술적인 역량뿐만 아니라, 오픈소스 프로젝트에서의 커뮤니케이션의 중요성을 깊이 체감할 수 있었습니다. 단순히 코드를 작성하고 제출하는 것을 넘어, Maintainer 및 다른 기여자들과의 지속적인 소통, 피드백 수용, 그리고 협업 방향 조율이 얼마나 중요한지 직접 경험하게 되었습니다. 특히 이슈를 공유하고, 작업 계획을 명확히 전달하며, 리뷰어의 피드백을 빠르게 반영하는 과정을 통해 오픈소스 생태계의 협업 문화가 신뢰와 효율을 만들어내는 구조임을 배웠습니다. 첫 번째 PR을 통해 긍정적인 피드백을 받은 뒤, Maintainer로부터 요청받았던 Flink 1.20 및 1.19 버전에 대한 backport 작업 역시 빠르게 진행하고 싶다는 동기부여가 생겼고, 기술적 기여뿐 아니라 프로젝트 요구에 신속히 대응하는 태도의 중요성도 느꼈습니다. 또한, 오픈소스 멘토링에서 운영진 역할을 맡아 저의 기여를 하면서 동시에 많은 멘티들이 오픈소스에 도전하는 과정을 지켜보고, 그분들께 답글로 방향성을 제시하는 경험은 저 자신의 성장에도 큰 자극이 되었습니다. 비록 오늘 포스팅에서는 멘토링 내용이 자세히 다뤄지지 않았지만, 저는 앞으로도 이 멘토링 활동과 오픈소스 기여를 꾸준히 이어갈 것이며, 우리나라에 올바르고 건강한 오픈소스 기여 문화가 자리잡도록 노력하는 1세대가 되고자 합니다. 앞으로도 단순한 코드 제공을 넘어, 신뢰를 바탕으로 커뮤니케이션하고, 프로젝트에 실질적인 가치를 더하는 기여자가 되기 위해 꾸준히 노력하겠습니다.
🖊️ 오픈소스 기여 · 2025-05-23
📘 Kafka RetryTopic 관련 기본 Template Bean 이름 변경
✏️ 1. 서론 개발자로서 오픈소스에 기여하는 일은 언젠가 꼭 해보고 싶은 목표 중 하나였습니다. 하지만 막상 시도하려고 하면, 어디서부터 시작해야 할지 막막한 것이 현실이었습니다. 그러던 중, 오픈소스 기여 방법을 안내해주는 커뮤니티를 알게 되었고, 그곳에서 멘토링 프로그램에 참여하게 되었습니다. 이슈를 찾고, 기여할 부분을 고민하며, 본격적으로 오픈소스의 세계에 발을 들이게 되었죠. Kafka를 공부하던 어느 날, 문득 Spring Kafka의 내부 구현이 궁금해졌고, 자연스럽게 GitHub 저장소를 들여다보게 되었습니다. 거기서 발견한 하나의 이슈 — 작아 보이지만 사용자 혼란을 야기할 수 있는 문제 — 를 계기로 생애 첫 Pull Request를 만들게 되었고, 운 좋게도 머지되며 Spring Kafka의 공식 Contributor가 될 수 있었습니다. 이 글에서는 그 여정을 공유하며, 어떤 이슈를 발견하고 어떻게 기여했는지를 정리해보려 합니다. 오픈소스 기여를 꿈꾸지만 막연한 분들께 작게나마 도움이 되었으면 합니다. ◈ ✏️ 2. 본론 🤔 2.1. 이슈 선택 과정 이슈를 선정할 때 가장 중요하게 생각한 기준은 기여가 비교적 쉬운가?, 그리고 해당 프로젝트의 소스 코드를 깊이 이해하지 않더라도 수정이 가능한가? 였습니다. 이러한 기준에 따라 Apache 프로젝트, Spring 프로젝트 등 여러 오픈소스 저장소를 둘러보며 적절한 이슈를 찾기 시작했습니다. 처음에는 Apache Jackrabbit Oak 프로젝트에서 기여할 만한 이슈를 찾을 수 있었습니다. 자세한 내용은 뒤에서 설명하겠지만, 결과적으로 해당 PR은 머지되지 못했고, 다시 새로운 이슈를 찾아 나섰습니다. 그러던 중, Spring-Kafka 에서 KafkaTemplate Bean 이름 불일치 이슈 를 발견하게 되었고, 그 내용을 바탕으로 기여하게 되었습니다. 📚 2.2. 이슈: KafkaTemplate Bean 이름 불일치 Spring Kafka의 공식 문서에서는 @RetryableTopic 이 기본적으로 사용할 KafkaTemplate 빈의 이름을 defaultRetryTopicKafkaTemplate라고 명시하고 있습니다. 그러나 실제 @RetryableTopic 의 JavaDoc에서는 다음과 같이 설명하고 있었습니다. * If not specified, a bean with name {@code retryTopicDefaultKafkaTemplate} or {@code kafkaTemplate} will be looked up. 즉, JavaDoc에는 retryTopicDefaultKafkaTemplate, 공식 문서에는 defaultRetryTopicKafkaTemplate라는 이름이 나와 있었고, 테스트 코드 일부도 JavaDoc 쪽 이름을 따라가고 있었습니다. 이로 인해 어떤 빈 이름을 사용해야 하는지 명확하지 않았습니다. 👉 해당 이슈는 Spring Kafka GitHub Issue #3514 에서 논의되었습니다. ‍💻 2.3. PR: JavaDoc 정정 먼저, @RetryableTopic 어노테이션 클래스의 JavaDoc을 공식 문서와 일치하도록 수정했습니다. Before: * The bean name of the {@link org.springframework.kafka.core.KafkaTemplate} bean that * will be used to forward the message to the retry and Dlt topics. If not specified, * a bean with name {@code retryTopicDefaultKafkaTemplate} or {@code kafkaTemplate} * will be looked up. * * @return the kafkaTemplate bean name. After: * The bean name of the {@link org.springframework.kafka.core.KafkaTemplate} bean that * will be used to forward the message to the retry and Dlt topics. If not specified, * a bean with name {@code defaultRetryTopicKafkaTemplate} or {@code kafkaTemplate} * will be looked up. * * @return the kafkaTemplate bean name. bean 으로 retryTopicDefaultKafkaTemplate 대신 defaultRetryTopicKafkaTemplate 사용한다고 수정을 하였습니다. 이 수정으로 공식 문서와 JavaDoc이 일치하게 되었고, 실제 사용자들이 혼란 없이 올바른 빈 이름을 사용할 수 있게 되었습니다. 👉 Commit Message → Fix: Replace retryTopicDefaultKafkaTemplate with defaultRetryTopicKafkaTemplate in docs 💻 2.4. PR: 테스트 코드 정정 또한 테스트 코드 내에서도 잘못된 빈 이름 문자열을 직접 명시하고 있었습니다. RetryTopicConfigurationProviderTests 라는 테스트 클래스에서 사용 중이었는데, 이 부분은 상수 RetryTopicBeanNames.DEFAULT_KAFKA_TEMPLATE_BEAN_NAME 을 사용하도록 리팩터링했습니다. Before: @Test void shouldProvideFromAnnotation() { // setup willReturn(kafkaOperations).given(beanFactory).getBean("retryTopicDefaultKafkaTemplate", KafkaOperations.class); // given RetryTopicConfigurationProvider provider = new RetryTopicConfigurationProvider(beanFactory); RetryTopicConfiguration configuration = provider.findRetryConfigurationFor(topics, annotatedMethod, bean); RetryTopicConfiguration configurationFromClass = provider .findRetryConfigurationFor(topics, null, AnnotatedClass.class, bean); // then then(this.beanFactory).should(times(0)).getBeansOfType(RetryTopicConfiguration.class); assertThat(configuration).isNotNull(); assertThat(configurationFromClass).isNotNull(); } @Test void shouldProvideFromMetaAnnotation() { // setup willReturn(kafkaOperations).given(beanFactory).getBean("retryTopicDefaultKafkaTemplate", KafkaOperations.class); // given RetryTopicConfigurationProvider provider = new RetryTopicConfigurationProvider(beanFactory); RetryTopicConfiguration configuration = provider.findRetryConfigurationFor(topics, metaAnnotatedMethod, bean); RetryTopicConfiguration configurationFromClass = provider .findRetryConfigurationFor(topics, null, MetaAnnotatedClass.class, bean); // then then(this.beanFactory).should(times(0)).getBeansOfType(RetryTopicConfiguration.class); assertThat(configuration).isNotNull(); assertThat(configuration.getConcurrency()).isEqualTo(3); assertThat(configurationFromClass).isNotNull(); assertThat(configurationFromClass.getConcurrency()).isEqualTo(3); } After: @Test void shouldProvideFromAnnotation() { // setup willReturn(kafkaOperations).given(beanFactory).getBean(RetryTopicBeanNames.DEFAULT_KAFKA_TEMPLATE_BEAN_NAME, KafkaOperations.class); // given RetryTopicConfigurationProvider provider = new RetryTopicConfigurationProvider(beanFactory); RetryTopicConfiguration configuration = provider.findRetryConfigurationFor(topics, annotatedMethod, bean); RetryTopicConfiguration configurationFromClass = provider .findRetryConfigurationFor(topics, null, AnnotatedClass.class, bean); // then then(this.beanFactory).should(times(0)).getBeansOfType(RetryTopicConfiguration.class); assertThat(configuration).isNotNull(); assertThat(configurationFromClass).isNotNull(); } @Test void shouldProvideFromMetaAnnotation() { // setup willReturn(kafkaOperations).given(beanFactory).getBean(RetryTopicBeanNames.DEFAULT_KAFKA_TEMPLATE_BEAN_NAME, KafkaOperations.class); // given RetryTopicConfigurationProvider provider = new RetryTopicConfigurationProvider(beanFactory); RetryTopicConfiguration configuration = provider.findRetryConfigurationFor(topics, metaAnnotatedMethod, bean); RetryTopicConfiguration configurationFromClass = provider .findRetryConfigurationFor(topics, null, MetaAnnotatedClass.class, bean); // then then(this.beanFactory).should(times(0)).getBeansOfType(RetryTopicConfiguration.class); assertThat(configuration).isNotNull(); assertThat(configuration.getConcurrency()).isEqualTo(3); assertThat(configurationFromClass).isNotNull(); assertThat(configurationFromClass.getConcurrency()).isEqualTo(3); } "retryTopicDefaultKafkaTemplate" 으로 되어있던 Bean 이름 부분을 RetryTopicBeanNames.DEFAULT_KAFKA_TEMPLATE_BEAN_NAME 로 변경하였습니다. 이를 통해 빈 이름도 변경하였고, 하드코딩된 문자열을 제거하면서, 상수를 사용하는 방식으로 코드의 일관성과 안정성을 확보했습니다. 추후 KafkaTemplate 빈 이름이 변경되더라도 상수만 수정하면 되므로 유지보수가 용이해졌습니다. 👉 Commit Message → Fix: Replace retryTopicDefaultKafkaTemplate with RetryTopicBeanNames.DEFAULT_KAFKA_TEMPLATE_BEAN_NAME in Test Code 그 결과, Spring Kafka 의 Contributor 가 되었고, 제 생의 첫 오픈 소스 기여에 성공하게 되었습니다. 해당 PR 내용을 확인하고 싶다면, Spring Kafka PR 기여 성공 해당 링크를 클릭하시기 바랍니다. 🤣 2.5. 배움: Apache Jackrabbit Oak 기여 실패 경험 위에서 언급하였지만, Spring Kafka 를 기여하기 전에, Apache Jackrabbit Oak 의 이슈를 하나 먼저 찾을 수 있었습니다. 💡 Apache Jackrabbit Oak ? 파일이나 문서 같은 데이터를 트리 구조로 저장하고 관리해주는 시스템입니다. Java로 만들어졌고, Adobe Experience Manager(AEM) 같은 유명한 콘텐츠 관리 시스템에서 내부 저장소로 사용되고 있습니다. 예를 들면, 웹사이트에서 페이지를 만들거나 이미지를 업로드할 때, 그 데이터를 저장하는 데 Oak가 쓰일 수 있습니다. 파일이나 폴더를 디렉토리처럼 관리하고, 누가 언제 바꿨는지 기록도 남길 수 있어서, 문서 관리 시스템이나 CMS에 딱 맞는 구조입니다. 당시 이슈 내용은 다음과 같았는데 MAX_SEGMENT_SIZE 상수가 Segment와 SegmentDataUtil s에 중복 정의되어 있으니 하나로 공유 할 수 있게 해달라는 내용이었습니다. Segment 라는 클래스에서 static final int MAX_SEGMENT_SIZE = 1 << 18; 로 되어있는 부분에 public 을 달아주고, 그걸 SegmentDataUtil 에서 사용 할 수 있게 해주는게 전부였습니다. 그래서 아래와 같이, 코드를 수정하려 PR 을 보냈습니다. 하지만 결과는 Merge 실패였습니다. 이유는 간단했습니다. 이미 저보다 먼저 해당 이슈를 캐치하고 PR 을 올린 사람이 있었기 때문이었습니다. 오픈 소스 생태계를 보면, 처음에는 잘 몰라서 실수 할 수 있는 2가지가 있습니다. 하나는 완료 된 이슈인데도 이슈가 Close 되어있지 않은게 있을 수 있다는 것입니다. 또 다른 하나는 누군가 해당 이슈를 해결하기 위해 Assign 을 받았는데, 그것을 가로채는 행동을 해서는 안된다는 것입니다. 보통은 이슈에 댓글로 본인이 해결하겠다고 리플을 달아두고 진행을 합니다. 그렇기 때문에, 어떠한 이슈를 해결하고자 한다면, 꼭 댓글에 본인이 해결하겠다는 의사를 밝히고, 진행하는 것이 좋습니다. 그리고 다른 사람이 진행을 하겠다고 댓글을 이미 남긴 상태라면, 그걸 가로채서는 안됩니다. 다만, 댓글이 달리고 한참이 지났는데도 해결이 되지 않은 이슈라면, 해당 이슈가 현재 해결 중인지, 그게 아니라면 내가 해결을 해도 되는지 댓글로 물어보고 진행을 하는 방향도 있습니다. ◈ ✏️ 3. 결론 이번 기여를 통해 Spring Kafka 문서와 코드의 불일치 문제를 해결하며, 보다 명확하고 유지보수하기 쉬운 기반을 마련하는 데 작은 보탬이 될 수 있었습니다. 비록 작고 사소한 변경처럼 보일 수 있지만, 실제 사용자 경험에 직접적인 영향을 줄 수 있다는 점에서 그 의미는 결코 작지 않았습니다. 무엇보다 오픈소스 생태계의 문화, 프로세스, 협업 방식을 실제로 체험하고 배울 수 있었던 값진 경험이었습니다. 오픈소스 기여를 어렵게만 느끼는 분들도 많겠지만, 완벽하게 아는 상태에서 시작하는 사람은 없습니다. 저 역시 문서 하나 고치고, 테스트 코드 한 줄 바꾸는 데서 시작했습니다. 작은 기여도 충분히 의미 있고, 오픈소스 커뮤니티는 그런 기여를 소중히 여깁니다. 👨 “중요한 건 완벽함이 아니라 시작입니다. 지금 여러분도 첫 발을 내디뎌보세요.”
🖊️ 오픈소스 기여 · 2024-10-09

Touch background to close