CODEWITHSUNDEEP
datastaxapache-pulsarspring-pulsar
Marcus Hellberg
Marcus Hellberg

Reputation: 1923

Spring Pulsar listener does not receive messages

Edit: the issue was that my @PulsarListener was in a class that was not a Spring Bean, and as such did not get discovered. Adding a @Service annotation solved the problem.

I'm in a Spring Boot 3.0.1 project and using version 0.1.1-SNAPSHOT of org.springframework.pulsar:spring-pulsar-spring-boot-starter.

I'm on a arm64 Mac, but I'm running a x86 JDK (19.0.2-amzn) because conscrypt does not support arm64 yet.

I am able to publish messages successfully, but I'm not receiving any messages.

There are no exceptions in the log unless I turn on debug logging for org.apache.pulsar. With debug logging, I get the following exceptions:

ava.lang.ClassNotFoundException: org.apache.pulsar.shade.org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport
    at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:641) ~[na:na]
    at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:188) ~[na:na]
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521) ~[na:na]
    at java.base/java.lang.Class.forName0(Native Method) ~[na:na]
    at java.base/java.lang.Class.forName(Class.java:495) ~[na:na]
    at java.base/java.lang.Class.forName(Class.java:474) ~[na:na]
    at org.springframework.boot.devtools.restart.classloader.RestartClassLoader.loadClass(RestartClassLoader.java:121) ~[spring-boot-devtools-3.0.1.jar:3.0.1]
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521) ~[na:na]
    at java.base/java.lang.Class.forName0(Native Method) ~[na:na]
    at java.base/java.lang.Class.forName(Class.java:495) ~[na:na]
    at java.base/java.lang.Class.forName(Class.java:474) ~[na:na]
    at org.apache.pulsar.shade.org.apache.commons.lang3.ClassUtils.getClass(ClassUtils.java:1069) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.org.apache.commons.lang3.ClassUtils.getClass(ClassUtils.java:1135) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.org.apache.commons.lang3.ClassUtils.getClass(ClassUtils.java:1118) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.common.util.ObjectMapperFactory.setAnnotationsModule(ObjectMapperFactory.java:203) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.common.util.ObjectMapperFactory.create(ObjectMapperFactory.java:117) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.springframework.pulsar.autoconfigure.AuthParameterUtils.maybeConvertToEncodedParamString(AuthParameterUtils.java:65) ~[spring-pulsar-spring-boot-autoconfigure-0.1.1-20230128.101302-74.jar:0.1.1-SNAPSHOT]
    at org.springframework.boot.context.properties.PropertyMapper$Source.lambda$as$2(PropertyMapper.java:197) ~[spring-boot-3.0.1.jar:3.0.1]
    at org.springframework.boot.context.properties.PropertyMapper$Source.to(PropertyMapper.java:291) ~[spring-boot-3.0.1.jar:3.0.1]
    at org.springframework.pulsar.autoconfigure.PulsarProperties$Client.buildProperties(PulsarProperties.java:1616) ~[spring-pulsar-spring-boot-autoconfigure-0.1.1-20230128.101302-74.jar:0.1.1-SNAPSHOT]
    at org.springframework.pulsar.autoconfigure.PulsarProperties.buildClientProperties(PulsarProperties.java:113) ~[spring-pulsar-spring-boot-autoconfigure-0.1.1-20230128.101302-74.jar:0.1.1-SNAPSHOT]
    at org.springframework.pulsar.autoconfigure.PulsarAutoConfiguration.pulsarClientConfiguration(PulsarAutoConfiguration.java:75) ~[spring-pulsar-spring-boot-autoconfigure-0.1.1-20230128.101302-74.jar:0.1.1-SNAPSHOT]
    at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104) ~[na:na]
    at java.base/java.lang.reflect.Method.invoke(Method.java:578) ~[na:na]
    at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:139) ~[spring-beans-6.0.3.jar:6.0.3]
    at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:653) ~[spring-beans-6.0.3.jar:6.0.3]
    at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:491) ~[spring-beans-6.0.3.jar:6.0.3]
    [...]

java.lang.IllegalArgumentException: Failed to load any of the given libraries: [netty_tcnative_osx_x86_64, netty_tcnative_x86_64, netty_tcnative]
    at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.loadFirstAvailable(NativeLibraryLoader.java:114) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.handler.ssl.OpenSsl.loadTcNative(OpenSsl.java:705) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.handler.ssl.OpenSsl.<clinit>(OpenSsl.java:146) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.handler.ssl.SslContext.defaultProvider(SslContext.java:124) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.handler.ssl.SslContext.defaultClientProvider(SslContext.java:120) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.handler.ssl.SslContext.newClientContextInternal(SslContext.java:817) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.handler.ssl.SslContextBuilder.build(SslContextBuilder.java:611) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.common.util.SecurityUtility.createNettySslContextForClient(SecurityUtility.java:303) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.common.util.SecurityUtility.createNettySslContextForClient(SecurityUtility.java:287) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.common.util.SecurityUtility.createNettySslContextForClient(SecurityUtility.java:211) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.client.impl.PulsarChannelInitializer.lambda$new$0(PulsarChannelInitializer.java:122) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.client.util.ObjectCache.get(ObjectCache.java:48) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.client.impl.PulsarChannelInitializer.lambda$initTls$1(PulsarChannelInitializer.java:171) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:995) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at java.base/java.lang.Thread.run(Thread.java:1589) ~[na:na]
    Suppressed: java.lang.UnsatisfiedLinkError: could not load a native library: org_apache_pulsar_shade_netty_tcnative_osx_x86_64
        at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.load(NativeLibraryLoader.java:239) ~[pulsar-client-all-2.10.2.jar:2.10.2]
        at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.loadFirstAvailable(NativeLibraryLoader.java:105) ~[pulsar-client-all-2.10.2.jar:2.10.2]
        ... 20 common frames omitted
    Caused by: java.io.FileNotFoundException: META-INF/native/liborg_apache_pulsar_shade_netty_tcnative_osx_x86_64.jnilib
        at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.load(NativeLibraryLoader.java:181)
        ... 21 common frames omitted
        Suppressed: java.lang.UnsatisfiedLinkError: no org_apache_pulsar_shade_netty_tcnative_osx_x86_64 in java.library.path: /Users/mhellber/Library/Java/Extensions:/Library/Java/Extensions:/Network/Library/Java/Extensions:/System/Library/Java/Extensions:/usr/lib/java:.
            at java.base/java.lang.ClassLoader.loadLibrary(ClassLoader.java:2444)
            at java.base/java.lang.Runtime.loadLibrary0(Runtime.java:848)
            at java.base/java.lang.System.loadLibrary(System.java:2047)
            at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryUtil.loadLibrary(NativeLibraryUtil.java:38)
            at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.loadLibrary(NativeLibraryLoader.java:391)
            at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.load(NativeLibraryLoader.java:161)
            ... 21 common frames omitted
            Suppressed: java.lang.UnsatisfiedLinkError: no org_apache_pulsar_shade_netty_tcnative_osx_x86_64 in java.library.path: /Users/mhellber/Library/Java/Extensions:/Library/Java/Extensions:/Network/Library/Java/Extensions:/System/Library/Java/Extensions:/usr/lib/java:.
                at java.base/java.lang.ClassLoader.loadLibrary(ClassLoader.java:2444)
                at java.base/java.lang.Runtime.loadLibrary0(Runtime.java:848)
                at java.base/java.lang.System.loadLibrary(System.java:2047)
                at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryUtil.loadLibrary(NativeLibraryUtil.java:38)
                at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104)
                at java.base/java.lang.reflect.Method.invoke(Method.java:578)
                at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader$1.run(NativeLibraryLoader.java:425)
                at java.base/java.security.AccessController.doPrivileged(AccessController.java:318)
                at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.loadLibraryByHelper(NativeLibraryLoader.java:417)
                at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.loadLibrary(NativeLibraryLoader.java:383)
                ... 22 common frames omitted
    Suppressed: java.lang.UnsatisfiedLinkError: could not load a native library: org_apache_pulsar_shade_netty_tcnative_x86_64
        at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.load(NativeLibraryLoader.java:239) ~[pulsar-client-all-2.10.2.jar:2.10.2]
        at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.loadFirstAvailable(NativeLibraryLoader.java:105) ~[pulsar-client-all-2.10.2.jar:2.10.2]
        ... 20 common frames omitted
    Caused by: java.io.FileNotFoundException: META-INF/native/liborg_apache_pulsar_shade_netty_tcnative_x86_64.jnilib
        at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.load(NativeLibraryLoader.java:181)
        ... 21 common frames omitted
        Suppressed: java.lang.UnsatisfiedLinkError: no org_apache_pulsar_shade_netty_tcnative_x86_64 in java.library.path: /Users/mhellber/Library/Java/Extensions:/Library/Java/Extensions:/Network/Library/Java/Extensions:/System/Library/Java/Extensions:/usr/lib/java:.
            at java.base/java.lang.ClassLoader.loadLibrary(ClassLoader.java:2444)
            at java.base/java.lang.Runtime.loadLibrary0(Runtime.java:848)
            at java.base/java.lang.System.loadLibrary(System.java:2047)
            at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryUtil.loadLibrary(NativeLibraryUtil.java:38)
            at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.loadLibrary(NativeLibraryLoader.java:391)
            at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.load(NativeLibraryLoader.java:161)
            ... 21 common frames omitted
            Suppressed: java.lang.UnsatisfiedLinkError: no org_apache_pulsar_shade_netty_tcnative_x86_64 in java.library.path: /Users/mhellber/Library/Java/Extensions:/Library/Java/Extensions:/Network/Library/Java/Extensions:/System/Library/Java/Extensions:/usr/lib/java:.
                at java.base/java.lang.ClassLoader.loadLibrary(ClassLoader.java:2444)
                at java.base/java.lang.Runtime.loadLibrary0(Runtime.java:848)
                at java.base/java.lang.System.loadLibrary(System.java:2047)
                at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryUtil.loadLibrary(NativeLibraryUtil.java:38)
                at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104)
                at java.base/java.lang.reflect.Method.invoke(Method.java:578)
                at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader$1.run(NativeLibraryLoader.java:425)
                at java.base/java.security.AccessController.doPrivileged(AccessController.java:318)
                at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.loadLibraryByHelper(NativeLibraryLoader.java:417)
                at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.loadLibrary(NativeLibraryLoader.java:383)
                ... 22 common frames omitted
    Suppressed: java.lang.UnsatisfiedLinkError: could not load a native library: org_apache_pulsar_shade_netty_tcnative
        at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.load(NativeLibraryLoader.java:239) ~[pulsar-client-all-2.10.2.jar:2.10.2]
        at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.loadFirstAvailable(NativeLibraryLoader.java:105) ~[pulsar-client-all-2.10.2.jar:2.10.2]
        ... 20 common frames omitted
    Caused by: java.io.FileNotFoundException: META-INF/native/liborg_apache_pulsar_shade_netty_tcnative.jnilib
        at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.load(NativeLibraryLoader.java:181)
        ... 21 common frames omitted
        Suppressed: java.lang.UnsatisfiedLinkError: no org_apache_pulsar_shade_netty_tcnative in java.library.path: /Users/mhellber/Library/Java/Extensions:/Library/Java/Extensions:/Network/Library/Java/Extensions:/System/Library/Java/Extensions:/usr/lib/java:.
            at java.base/java.lang.ClassLoader.loadLibrary(ClassLoader.java:2444)
            at java.base/java.lang.Runtime.loadLibrary0(Runtime.java:848)
            at java.base/java.lang.System.loadLibrary(System.java:2047)
            at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryUtil.loadLibrary(NativeLibraryUtil.java:38)
            at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.loadLibrary(NativeLibraryLoader.java:391)
            at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.load(NativeLibraryLoader.java:161)
            ... 21 common frames omitted
            Suppressed: java.lang.UnsatisfiedLinkError: no org_apache_pulsar_shade_netty_tcnative in java.library.path: /Users/mhellber/Library/Java/Extensions:/Library/Java/Extensions:/Network/Library/Java/Extensions:/System/Library/Java/Extensions:/usr/lib/java:.
                at java.base/java.lang.ClassLoader.loadLibrary(ClassLoader.java:2444)
                at java.base/java.lang.Runtime.loadLibrary0(Runtime.java:848)
                at java.base/java.lang.System.loadLibrary(System.java:2047)
                at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryUtil.loadLibrary(NativeLibraryUtil.java:38)
                at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104)
                at java.base/java.lang.reflect.Method.invoke(Method.java:578)
                at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader$1.run(NativeLibraryLoader.java:425)
                at java.base/java.security.AccessController.doPrivileged(AccessController.java:318)
                at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.loadLibraryByHelper(NativeLibraryLoader.java:417)
                at org.apache.pulsar.shade.io.netty.util.internal.NativeLibraryLoader.loadLibrary(NativeLibraryLoader.java:383)
                ... 22 common frames omitted

java.lang.UnsatisfiedLinkError: 'int org.apache.pulsar.shade.io.netty.internal.tcnative.Library.aprMajorVersion()'
    at org.apache.pulsar.shade.io.netty.internal.tcnative.Library.aprMajorVersion(Native Method) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.internal.tcnative.Library.initialize(Library.java:191) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.handler.ssl.OpenSsl.initializeTcNative(OpenSsl.java:710) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.handler.ssl.OpenSsl.<clinit>(OpenSsl.java:163) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.handler.ssl.SslContext.defaultProvider(SslContext.java:124) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.handler.ssl.SslContext.defaultClientProvider(SslContext.java:120) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.handler.ssl.SslContext.newClientContextInternal(SslContext.java:817) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.handler.ssl.SslContextBuilder.build(SslContextBuilder.java:611) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.common.util.SecurityUtility.createNettySslContextForClient(SecurityUtility.java:303) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.common.util.SecurityUtility.createNettySslContextForClient(SecurityUtility.java:287) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.common.util.SecurityUtility.createNettySslContextForClient(SecurityUtility.java:211) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.client.impl.PulsarChannelInitializer.lambda$new$0(PulsarChannelInitializer.java:122) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.client.util.ObjectCache.get(ObjectCache.java:48) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.client.impl.PulsarChannelInitializer.lambda$initTls$1(PulsarChannelInitializer.java:171) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:995) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at org.apache.pulsar.shade.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[pulsar-client-all-2.10.2.jar:2.10.2]
    at java.base/java.lang.Thread.run(Thread.java:1589) ~[na:na]

I tried adding netty-tcnative as a dependency in my pom.xml, but it did not make a difference.

Upvotes: 1

Views: 1146

Answers (1)

onobc
onobc

Reputation: 611

All of these exceptions are red herrings - when level is DEBUG this is expected from Pulsar.

The netty-tcnative logs - this is just part of its load process. It can't find it so security defaults to JDK provider. Note the DEBUG level of the following statement:

2023-01-28T20:47:16.894-06:00 DEBUG 19255 --- [r-client-io-1-1] o.a.p.s.io.netty.handler.ssl.OpenSsl     : Failed to load netty-tcnative; OpenSslEngine will be unavailable, unless the application has already loaded the symbols by some other means. See https://netty.io/wiki/forked-tomcat-native.html for more information.

And the CNFEX is in the pulsar code and expected (see ObjectMapperFactory.

I created a sample app that sends and receives using:

  • SpringBoot 3.0.2
  • Java 19 (x86 JDK (19.0.2-amzn)
  • Spring Pulsar 0.1.1-SNAPSHOT

There is something else going on w/ your app. Turn apache logging back to ERROR and try this simple app:

@SpringBootApplication
public class TcnativeApplication {

    public static void main(String[] args) {
        SpringApplication.run(TcnativeApplication.class, args);
    }

    @Bean
    ApplicationRunner runner1(PulsarTemplate<String> pulsarTemplate) {
        String topic1 = "your-topic-goes-here";
        return args -> {
            for (int i = 0; i < 10; i++) {
                pulsarTemplate.send(topic1, "This is message " + (i + 1));
            }
        };
    }

    @PulsarListener(subscriptionName = "subscription-1", topics = "your-topic-goes-here")
    void listen1(String message) {
        System.out.println("***** " + message);
    }
}

About the @PulsarListener needing to be a @Service... ultimately the framework uses a Spring BeanPostProcessor(PulsarListenerAnnotationBeanPostProcessor) and any beans in the app context that are marked w/ @PulsarListener are processed accordingly. So @Bean, @Service or @Component are common ways to do that. The most common is to put the listener as a @Bean in a @Configuration class.

Upvotes: 3

Related Questions