Hi, I try to use RabbitMQ as a Source. My source consumes messages of the queue but the statefun is not execution - not even created.
This is my main function: 1 public static void main(String[] args) throws Exception { 2 3 final var env = StreamExecutionEnvironment.getExecutionEnvironment(); 4 5 env.registerTypeWithKryoSerializer(Any.class, ProtobufSerializer.class); 6 7 env.enableCheckpointing(10000, CheckpointingMode.EXACTLY_ONCE); 8 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000); 9 env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); 10 env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); 11 12 final var statefunConfig = StatefulFunctionsConfig.fromEnvironment(env); 13 statefunConfig.setFlinkJobName("test"); 14 statefunConfig.setFactoryType(MessageFactoryType.WITH_KRYO_PAYLOADS); 15 16 final var connectionConfig = new RMQConnectionConfig.Builder() 17 .setHost("localhost") 18 .setUserName("guest") 19 .setPassword("guest") 20 .setPort(5672) 21 .setVirtualHost("test") 22 .setPrefetchCount(5000) 23 .build(); 24 25 final var deserializationSchema = new TypeInformationSerializationSchema<>( 26 new ProtobufTypeInformation<>(Any.class), env.getConfig()); 27 final var rmqSource = new RMQSource<>(connectionConfig, TEST_INGRESS, true, deserializationSchema); 28 29 final var source = env 30 .addSource(rmqSource, TEST_INGRESS) 31 .setParallelism(1) 32 .map(msg -> { 33 return RoutableMessageBuilder 34 .builder() 35 .withTargetAddress(MyStatefun.TYPE, Utils.getUUID()) 36 .withMessageBody(msg) 37 .build(); 38 }); 39 40 StatefulFunctionDataStreamBuilder 41 .builder("test") 42 .withDataStreamAsIngress(source) 43 .withFunctionProvider(MyStatefun.TYPE, unused -> { 44 return new MyStatefun(); 45 }) 46 .withEgressId(MyStatefun.EGRESS) 47 .withConfiguration(statefunConfig) 48 .build(env) 49 .getDataStreamForEgressId(MyStatefun.EGRESS) 50 .addSink(new PrintSinkFunction<>(true)); 51 52 env.execute(); 53 54 } A breakpoint in line 33 shows me the messages consumed. A breakpoint in line 44 is never called. The message is reportingly consumed but never acknowledged or processed. Before using RabbitMQ I used a custom SourceFunction to fake input data and it worked well. To setup things I use a local environment but logging does not show up any errors. Before my current problem I had another error during message deserialization and it wasn't reported either. Unfortunately I didn't manage to get the exception in the log/stdout. I had to use the debugger to find the reason of the former problem. In this situation now the debugger shows no thrown or caught exceptions. That's way I stuck. Of course I would like to know what's the problem with my code. But I guess it is not obviously. Maybe some can give me a hint how to turn on exception logging which might help to get closer to the origin of the phenomenon. Thanks in advance, Stephan