@Override
    public void onMsg(TbContext ctx, TbMsg msg) {
        String topic = TbNodeUtils.processPattern(config.getTopicPattern(), msg);
        String keyPattern = config.getKeyPattern();
        var tbMsg = ackIfNeeded(ctx, msg);
        try {
            if (initError != null) {
                ctx.tellFailure(tbMsg, new RuntimeException("Failed to initialize Kafka rule node producer: " + initError.getMessage()));
            } else {
                ctx.getExternalCallExecutor().executeAsync(() -> {
                    publish(
                            ctx,
                            tbMsg,
                            topic,
                            keyPattern == null || keyPattern.isEmpty()
                                    ? null
                                    : TbNodeUtils.processPattern(config.getKeyPattern(), tbMsg)
                    );
                    return null;
                });
            }
        } catch (Exception e) {
            ctx.tellFailure(tbMsg, e);
        }
    }
@Test
    public void givenInitErrorIsNotNull_whenOnMsg_thenTellFailure() {
        // GIVEN
        String errorMsg = "Error during kafka initialization!";
        ReflectionTestUtils.setField(node, "config", config);
        ReflectionTestUtils.setField(node, "initError", new ThingsboardKafkaClientError(errorMsg));

        // WHEN
        TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT);
        node.onMsg(ctxMock, msg);

        // THEN
        ArgumentCaptor<Throwable> actualError = ArgumentCaptor.forClass(Throwable.class);
        then(ctxMock).should().tellFailure(eq(msg), actualError.capture());
        assertThat(actualError.getValue())
                .isInstanceOf(RuntimeException.class)
                .hasMessage("Failed to initialize Kafka rule node producer: " + errorMsg);
    }