<div id="readability-page-1" class="page"><div>이관 받은 코드에서 reactor-kafka 버전을 1.2.5.RELEASE 버전을 쓰고 있었는데, 버전 업그레이드를 하면서 겪은 이슈는 다음과 같았다.
<pre><code>2023-04-10 18:17:36.601 ERROR 1 --- [ard.processor-1] r.k.r.internals.ConsumerEventLoop : Unexpected exception 



java.lang.IllegalStateException: No subscriptions have been created 

at reactor.kafka.receiver.ReceiverOptions.subscriber(ReceiverOptions.java:515) 

at reactor.kafka.receiver.internals.ConsumerEventLoop$SubscribeEvent.run(ConsumerEventLoop.java:241) 

at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68) 

at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28) 

at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) 

at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) 

at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) 

at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)</code></pre>
왜 구독하지 못하는 것일까? 궁금한 부분이었는데, 특정 부분에서 버전에 따라 토피 정보가 설정 되고, 1.3.2 버전에서는 설정이 안된다.
<pre><code> final ReceiverOptions&lt;String, V&gt; options = ReceiverOptions.create(props);
 options.subscription(Collections.singleton(topicName));</code></pre>
위와 같은 코드였는데, 처음에 문제의 원인은 <code>Collections.singletonList</code>로 바꿔보자 였는데, 동일하게 안됐다. Reactor의 개념이 불변이라, 설정값 또한 인스턴스가 새로 생성 되기 때문으로 보인다.
<code>Collections.singleton</code> or <code>Collections.singletonList</code>를 쓰는 것은 해당 이슈와 무관하다.
추측한 부분과 새로운 버전의 차이를 알고보니 원인은 이렇다.
1.2.7 이하에는 ReceiverOptions 인터페이스를 구현하는 구현체는 총 두개다.
<ul data-ke-list-type="disc">
<li>MutableReceiverOptions
<ul data-ke-list-type="disc">
<li>Configuration properties for Reactive Kafka KafkaReceiver and its underlying KafkaConsumer.</li>
<li>deprecated in favor of ImmutableReceiverOptions and will be deleted in 3.x version</li>
</ul>
</li>
<li>ImmutableReceiverOptions</li>
</ul>
결론적으로 1.3.0 이후부터 MutableReceiverOptions 해당 구현체가 삭제 됐다.
as-is
<pre><code> final ReceiverOptions&lt;String, V&gt; options = ReceiverOptions.create(props);
 options.subscription(Collections.singleton(topicName));</code></pre>
to-be
<pre><code> ReceiverOptions&lt;String, V&gt; receiverOptions = ReceiverOptions.create(props);
 ReceiverOptions&lt;String, V&gt; options = receiverOptions.subscription(Collections.singleton(topicName));
 options.withValueDeserializer(valueDeserializer);</code></pre>
이렇게 코드를 변경 해주면 정상적으로 잘 된다.</div></div>

Seungdols Company

seungdols

이관 받은 코드에서 reactor-kafka 버전을 1.2.5.RELEASE 버전을 쓰고 있었는데, 버전 업그레이드를 하면서 겪은 이슈는 다음과 같았다. 2023-04-10 18:17:36.601 ERROR 1 --- [ard.processor-1] r.k.r.internals.ConsumerEventLoop : Unexpected exception java.lang.IllegalStateException: No subscriptions have b..

reactor-kafka version up 이슈 (No subscriptions have been created)

댓글