@InvokeOnHeader(CONTROL_ACTION_SUBSCRIBE)
    public void performSubscribe(final Message message, AsyncCallback callback) {
        String filterId;
        if (message.getBody() instanceof DynamicRouterControlMessage) {
            filterId = subscribeFromMessage(dynamicRouterControlService, message, false);
        } else {
            filterId = subscribeFromHeaders(dynamicRouterControlService, message, false);
        }
        message.setBody(filterId);
        callback.done(false);
    }
@Test
    void performSubscribeActionWithMessageInBodyWithEmptyExpression() {
        String subscribeChannel = "testChannel";
        DynamicRouterControlMessage subMsg = DynamicRouterControlMessage.Builder.newBuilder()
                .subscribeChannel(subscribeChannel)
                .subscriptionId("testId")
                .destinationUri("mock://test")
                .priority(10)
                .predicate("")
                .expressionLanguage("simple")
                .build();
        when(message.getBody()).thenReturn(subMsg);
        when(message.getBody(DynamicRouterControlMessage.class)).thenReturn(subMsg);
        Exception ex = assertThrows(IllegalStateException.class, () -> producer.performSubscribe(message, callback));
        assertEquals("Predicate bean could not be found", ex.getMessage());
    }