From da78d934680059f751b1869998485c0ebc551d6d Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Tue, 18 Jun 2024 17:59:57 +0300 Subject: [PATCH 1/4] added tests for the kafka node --- .../rule/engine/kafka/TbKafkaNode.java | 45 ++- .../rule/engine/kafka/TbKafkaNodeTest.java | 365 ++++++++++++++++++ 2 files changed, 390 insertions(+), 20 deletions(-) create mode 100644 rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java index fa07e7c06d..92f999450e 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java @@ -81,26 +81,7 @@ public class TbKafkaNode extends TbAbstractExternalNode { super.init(ctx); this.config = TbNodeUtils.convert(configuration, TbKafkaNodeConfiguration.class); this.initError = null; - Properties properties = new Properties(); - properties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-tb-kafka-node-" + ctx.getSelfId().getId().toString() + "-" + ctx.getServiceId()); - properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers()); - properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, config.getValueSerializer()); - properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, config.getKeySerializer()); - properties.put(ProducerConfig.ACKS_CONFIG, config.getAcks()); - properties.put(ProducerConfig.RETRIES_CONFIG, config.getRetries()); - properties.put(ProducerConfig.BATCH_SIZE_CONFIG, config.getBatchSize()); - properties.put(ProducerConfig.LINGER_MS_CONFIG, config.getLinger()); - properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, config.getBufferMemory()); - if (config.getOtherProperties() != null) { - config.getOtherProperties().forEach((k, v) -> { - if (SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG.equals(k) - || SslConfigs.SSL_KEYSTORE_KEY_CONFIG.equals(k) - || SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG.equals(k)) { - v = v.replace("\\n", "\n"); - } - properties.put(k, v); - }); - } + Properties properties = getKafkaProperties(ctx); addMetadataKeyValuesAsKafkaHeaders = BooleanUtils.toBooleanDefaultIfNull(config.isAddMetadataKeyValuesAsKafkaHeaders(), false); toBytesCharset = config.getKafkaHeadersCharset() != null ? Charset.forName(config.getKafkaHeadersCharset()) : StandardCharsets.UTF_8; try { @@ -160,6 +141,30 @@ public class TbKafkaNode extends TbAbstractExternalNode { } } + protected Properties getKafkaProperties(TbContext ctx) { + Properties properties = new Properties(); + properties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-tb-kafka-node-" + ctx.getSelfId().getId().toString() + "-" + ctx.getServiceId()); + properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers()); + properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, config.getValueSerializer()); + properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, config.getKeySerializer()); + properties.put(ProducerConfig.ACKS_CONFIG, config.getAcks()); + properties.put(ProducerConfig.RETRIES_CONFIG, config.getRetries()); + properties.put(ProducerConfig.BATCH_SIZE_CONFIG, config.getBatchSize()); + properties.put(ProducerConfig.LINGER_MS_CONFIG, config.getLinger()); + properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, config.getBufferMemory()); + if (config.getOtherProperties() != null) { + config.getOtherProperties().forEach((k, v) -> { + if (SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG.equals(k) + || SslConfigs.SSL_KEYSTORE_KEY_CONFIG.equals(k) + || SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG.equals(k)) { + v = v.replace("\\n", "\n"); + } + properties.put(k, v); + }); + } + return properties; + } + @Override public void destroy() { if (this.producer != null) { diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java new file mode 100644 index 0000000000..86c39336f4 --- /dev/null +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java @@ -0,0 +1,365 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.kafka; + +import org.apache.kafka.clients.producer.Callback; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeader; +import org.apache.kafka.common.header.internals.RecordHeaders; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.NullAndEmptySource; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.test.util.ReflectionTestUtils; +import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.common.util.ListeningExecutor; +import org.thingsboard.rule.engine.TestDbCallbackExecutor; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.rule.engine.api.TbNodeConfiguration; +import org.thingsboard.rule.engine.api.util.TbNodeUtils; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.RuleNodeId; +import org.thingsboard.server.common.data.msg.TbMsgType; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.TbMsgMetaData; + +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; +import java.util.Map; +import java.util.Properties; +import java.util.UUID; +import java.util.concurrent.Callable; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatNoException; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.then; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.BDDMockito.willThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; + +@ExtendWith(MockitoExtension.class) +public class TbKafkaNodeTest { + + private final DeviceId DEVICE_ID = new DeviceId(UUID.fromString("5f2eac08-bd1f-4635-a6c2-437369f996cf")); + private final ListeningExecutor executor = new TestDbCallbackExecutor(); + + private TbKafkaNode node; + private TbKafkaNodeConfiguration config; + + @Mock + private TbContext ctxMock; + @Mock + private Producer producerMock; + @Mock + private RecordMetadata recordMetadataMock; + + @BeforeEach + void setUp() { + node = new TbKafkaNode(); + config = new TbKafkaNodeConfiguration().defaultConfiguration(); + } + + @Test + public void verifyDefaultConfig() { + assertThat(config.getTopicPattern()).isEqualTo("my-topic"); + assertThat(config.getKeyPattern()).isNull(); + assertThat(config.getBootstrapServers()).isEqualTo("localhost:9092"); + assertThat(config.getRetries()).isEqualTo(0); + assertThat(config.getBatchSize()).isEqualTo(16384); + assertThat(config.getLinger()).isEqualTo(0); + assertThat(config.getBufferMemory()).isEqualTo(33554432); + assertThat(config.getAcks()).isEqualTo("-1"); + assertThat(config.getKeySerializer()).isEqualTo(StringSerializer.class.getName()); + assertThat(config.getValueSerializer()).isEqualTo(StringSerializer.class.getName()); + assertThat(config.getOtherProperties()).isEmpty(); + assertThat(config.isAddMetadataKeyValuesAsKafkaHeaders()).isFalse(); + assertThat(config.getKafkaHeadersCharset()).isEqualTo("UTF-8"); + } + + @Test + public void givenAddMetadataKeyValuesAsKafkaHeadersIsTrueAndKafkaHeadersCharsetIsSet_whenInit_thenOk() { + config.setAddMetadataKeyValuesAsKafkaHeaders(true); + config.setKafkaHeadersCharset("UTF-16"); + + String ruleNodeIdStr = "0d35733c-7661-4797-819e-d9188974e3b2"; + String serviceIdStr = "test-service"; + + given(ctxMock.getSelfId()).willReturn(new RuleNodeId(UUID.fromString(ruleNodeIdStr))); + given(ctxMock.getServiceId()).willReturn(serviceIdStr); + + assertThatNoException().isThrownBy(() -> node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)))); + + Boolean addMetadataKeyValuesAsKafkaHeaders = (Boolean) ReflectionTestUtils.getField(node, "addMetadataKeyValuesAsKafkaHeaders"); + Charset toBytesCharset = (Charset) ReflectionTestUtils.getField(node, "toBytesCharset"); + + assertThat(addMetadataKeyValuesAsKafkaHeaders).isTrue(); + assertThat(toBytesCharset).isEqualTo(StandardCharsets.UTF_16); + } + + @Test + public void verifyGetKafkaPropertiesMethod() { + String sslKeyStoreCertificateChain = "cbdvch\\nfwrg\nvgwg\\n"; + String sslKeyStoreKey = "nghmh\\nhmmnh\\\\ngreg\nvgwg\\n"; + String sslTruststoreCertificates = "grthrt\fd\\nfwrg\nvgwg\\n"; + config.setOtherProperties(Map.of( + "ssl.keystore.certificate.chain", sslKeyStoreCertificateChain, + "ssl.keystore.key", sslKeyStoreKey, + "ssl.truststore.certificates", sslTruststoreCertificates, + "ssl.protocol", "TLSv1.2" + )); + ReflectionTestUtils.setField(node, "config", config); + + String ruleNodeIdStr = "e646b885-8004-45b4-8bfb-78db21870e0f"; + String serviceIdStr = "test-service"; + given(ctxMock.getSelfId()).willReturn(new RuleNodeId(UUID.fromString(ruleNodeIdStr))); + given(ctxMock.getServiceId()).willReturn(serviceIdStr); + + Properties expectedProperties = new Properties(); + expectedProperties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-tb-kafka-node-" + ruleNodeIdStr + "-" + serviceIdStr); + expectedProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers()); + expectedProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, config.getValueSerializer()); + expectedProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, config.getKeySerializer()); + expectedProperties.put(ProducerConfig.ACKS_CONFIG, config.getAcks()); + expectedProperties.put(ProducerConfig.RETRIES_CONFIG, config.getRetries()); + expectedProperties.put(ProducerConfig.BATCH_SIZE_CONFIG, config.getBatchSize()); + expectedProperties.put(ProducerConfig.LINGER_MS_CONFIG, config.getLinger()); + expectedProperties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, config.getBufferMemory()); + expectedProperties.put("ssl.keystore.certificate.chain", sslKeyStoreCertificateChain.replace("\\n", "\n")); + expectedProperties.put("ssl.keystore.key", sslKeyStoreKey.replace("\\n", "\n")); + expectedProperties.put("ssl.truststore.certificates", sslTruststoreCertificates.replace("\\n", "\n")); + expectedProperties.put("ssl.protocol", "TLSv1.2"); + + Properties actualsProperties = node.getKafkaProperties(ctxMock); + assertThat(actualsProperties).isEqualTo(expectedProperties); + } + + @Test + public void givenInitErrorIsNotNull_whenOnMsg_thenTellFailure() { + init(); + String errorMsg = "Error during init!"; + ReflectionTestUtils.setField(node, "initError", new RuntimeException(errorMsg)); + + TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); + node.onMsg(ctxMock, msg); + + ArgumentCaptor 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); + } + + @Test + public void givenForceAckIsTrueAndExceptionWasThrown_whenOnMsg_thenTellFailure() { + init(); + ReflectionTestUtils.setField(node, "forceAck", true); + + ListeningExecutor executorMock = mock(ListeningExecutor.class); + given(ctxMock.getExternalCallExecutor()).willReturn(executorMock); + String errorMsg = "Something went wrong!"; + willThrow(new RuntimeException(errorMsg)).given(executorMock).executeAsync(any(Callable.class)); + + TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); + node.onMsg(ctxMock, msg); + + then(ctxMock).should().ack(msg); + ArgumentCaptor actualMsg = ArgumentCaptor.forClass(TbMsg.class); + ArgumentCaptor actualError = ArgumentCaptor.forClass(Throwable.class); + then(ctxMock).should().tellFailure(actualMsg.capture(), actualError.capture()); + assertThat(actualMsg.getValue()).usingRecursiveComparison().ignoringFields("ctx").isEqualTo(msg); + assertThat(actualError.getValue()).isInstanceOf(RuntimeException.class).hasMessage(errorMsg); + } + + @ParameterizedTest + @MethodSource + public void givenTopicAndKeyPatternsAndAddMetadataKeyValuesAsKafkaHeadersIsFalse_whenOnMsg_thenTellSuccess + (String topicPattern, String keyPattern, TbMsgMetaData metaData, String data) { + config.setTopicPattern(topicPattern); + config.setKeyPattern(keyPattern); + init(); + + TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, metaData, data); + String topic = TbNodeUtils.processPattern(topicPattern, msg); + String key = TbNodeUtils.processPattern(keyPattern, msg); + long offset = 1; + int partition = 0; + mockSuccessfulPublishingRequest(topic, offset, partition); + + node.onMsg(ctxMock, msg); + + verifyProducerRecord(topic, key, msg.getData()); + verifyOutboundMsg(offset, partition, topic, msg); + } + + private static Stream givenTopicAndKeyPatternsAndAddMetadataKeyValuesAsKafkaHeadersIsFalse_whenOnMsg_thenTellSuccess() { + return Stream.of( + Arguments.of("test-topic", "test-key", new TbMsgMetaData(), TbMsg.EMPTY_JSON_OBJECT), + Arguments.of("${mdTopicPattern}", "${mdKeyPattern}", new TbMsgMetaData( + Map.of( + "mdTopicPattern", "md-test-topic", + "mdKeyPattern", "md-test-key" + )), TbMsg.EMPTY_JSON_OBJECT), + Arguments.of("$[msgTopicPattern]", "$[msgKeyPattern]", new TbMsgMetaData(), + "{\"msgTopicPattern\":\"msg-test-topic\",\"msgKeyPattern\":\"msg-test-key\"}") + ); + } + + @Test + public void givenForceAckIsFalseAndAddMetadataKeyValuesAsKafkaHeadersIsTrueAndToBytesCharsetIsSet_whenOnMsg_thenAckAndTellSuccess() { + String topic = "test-topic"; + String key = "test-key"; + config.setTopicPattern(topic); + config.setKeyPattern(key); + config.setAddMetadataKeyValuesAsKafkaHeaders(true); + config.setKafkaHeadersCharset("UTF-16"); + init(); + ReflectionTestUtils.setField(node, "forceAck", false); + ReflectionTestUtils.setField(node, "addMetadataKeyValuesAsKafkaHeaders", true); + ReflectionTestUtils.setField(node, "toBytesCharset", StandardCharsets.UTF_16); + + TbMsgMetaData metaData = new TbMsgMetaData(); + metaData.putValue("key", "value"); + TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, metaData, TbMsg.EMPTY_JSON_OBJECT); + + long offset = 1; + int partition = 0; + + mockSuccessfulPublishingRequest(topic, offset, partition); + + node.onMsg(ctxMock, msg); + + then(ctxMock).should(never()).ack(msg); + Headers expectedHeaders = new RecordHeaders(); + msg.getMetaData().values().forEach((k, v) -> expectedHeaders.add(new RecordHeader("tb_msg_md_" + k, v.getBytes(StandardCharsets.UTF_16)))); + verifyProducerRecord(topic, key, msg.getData(), expectedHeaders); + verifyOutboundMsg(offset, partition, topic, msg); + } + + @ParameterizedTest + @NullAndEmptySource + public void givenKeyIsNullOrEmptyAndErrorOccursDuringPublishing_whenOnMsg_thenTellFailure(String key) { + String topic = "test-topic"; + config.setTopicPattern(topic); + config.setKeyPattern(key); + config.setAddMetadataKeyValuesAsKafkaHeaders(false); + + TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); + + String errorMsg = "Something went wrong!"; + + given(ctxMock.getExternalCallExecutor()).willReturn(executor); + willAnswer(invocation -> { + Callback callback = invocation.getArgument(1); + callback.onCompletion(recordMetadataMock, new RuntimeException(errorMsg)); + return null; + }).given(producerMock).send(any(), any(Callback.class)); + + init(); + node.onMsg(ctxMock, msg); + + verifyProducerRecord(topic, null, msg.getData()); + + ArgumentCaptor actualMsg = ArgumentCaptor.forClass(TbMsg.class); + ArgumentCaptor actualError = ArgumentCaptor.forClass(Throwable.class); + then(ctxMock).should().tellFailure(actualMsg.capture(), actualError.capture()); + TbMsgMetaData metaData = new TbMsgMetaData(); + metaData.putValue("error", RuntimeException.class + ": " + errorMsg); + TbMsg expectedMsg = TbMsg.transformMsgMetadata(msg, metaData); + assertThat(actualMsg.getValue()) + .usingRecursiveComparison() + .ignoringFields("ctx") + .isEqualTo(expectedMsg); + } + + @Test + public void givenProducerIsNotNull_whenDestroy_thenShouldClose() { + ReflectionTestUtils.setField(node, "producer", producerMock); + + node.destroy(); + + then(producerMock).should().close(); + } + + @Test + public void givenProducerIsNull_whenDestroy_thenDoNothing() { + node.destroy(); + then(producerMock).shouldHaveNoInteractions(); + } + + private void mockSuccessfulPublishingRequest(String topic, long offset, int partition) { + given(ctxMock.getExternalCallExecutor()).willReturn(executor); + willAnswer(invocation -> { + Callback callback = invocation.getArgument(1); + callback.onCompletion(recordMetadataMock, null); + return null; + }).given(producerMock).send(any(), any(Callback.class)); + given(recordMetadataMock.offset()).willReturn(offset); + given(recordMetadataMock.partition()).willReturn(partition); + given(recordMetadataMock.topic()).willReturn(topic); + } + + private void init() { + ReflectionTestUtils.setField(node, "config", config); + ReflectionTestUtils.setField(node, "producer", producerMock); + ReflectionTestUtils.setField(node, "addMetadataKeyValuesAsKafkaHeaders", false); + } + + private void verifyProducerRecord(String expectedTopic, String expectedKey, String expectedValue) { + verifyProducerRecord(expectedTopic, expectedKey, expectedValue, null); + } + + private void verifyProducerRecord(String expectedTopic, String expectedKey, String expectedValue, Headers expectedHeaders) { + ArgumentCaptor> actualRecordCaptor = ArgumentCaptor.forClass(ProducerRecord.class); + then(producerMock).should().send(actualRecordCaptor.capture(), any()); + ProducerRecord actualRecord = actualRecordCaptor.getValue(); + assertThat(actualRecord.topic()).isEqualTo(expectedTopic); + assertThat(actualRecord.key()).isEqualTo(expectedKey); + assertThat(actualRecord.value()).isEqualTo(expectedValue); + if (expectedHeaders != null) { + assertThat(actualRecord.headers()).isEqualTo(expectedHeaders); + } + } + + private void verifyOutboundMsg(long expectedOffset, long expectedPartition, String expectedTopic, TbMsg originalMsg) { + ArgumentCaptor actualMsg = ArgumentCaptor.forClass(TbMsg.class); + then(ctxMock).should().tellSuccess(actualMsg.capture()); + TbMsgMetaData metaData = originalMsg.getMetaData().copy(); + metaData.putValue("offset", String.valueOf(expectedOffset)); + metaData.putValue("partition", String.valueOf(expectedPartition)); + metaData.putValue("topic", expectedTopic); + TbMsg expectedMsg = TbMsg.transformMsgMetadata(originalMsg, metaData); + assertThat(actualMsg.getValue()) + .usingRecursiveComparison() + .ignoringFields("ctx") + .isEqualTo(expectedMsg); + } +} From 9bcfc75a773703386a9e716fd9b2aaacd30c6fcd Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Fri, 2 Aug 2024 15:11:28 +0300 Subject: [PATCH 2/4] reduced usage of reflection --- .../rule/engine/kafka/TbKafkaNode.java | 6 +- .../rule/engine/kafka/TbKafkaNodeTest.java | 280 +++++++++++------- 2 files changed, 175 insertions(+), 111 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java index 92f999450e..1b77e5c780 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java @@ -85,7 +85,7 @@ public class TbKafkaNode extends TbAbstractExternalNode { addMetadataKeyValuesAsKafkaHeaders = BooleanUtils.toBooleanDefaultIfNull(config.isAddMetadataKeyValuesAsKafkaHeaders(), false); toBytesCharset = config.getKafkaHeadersCharset() != null ? Charset.forName(config.getKafkaHeadersCharset()) : StandardCharsets.UTF_8; try { - this.producer = new KafkaProducer<>(properties); + this.producer = getKafkaProducer(properties); Thread ioThread = (Thread) ReflectionUtils.getField(IO_THREAD_FIELD, producer); ioThread.setUncaughtExceptionHandler((thread, throwable) -> { if (throwable instanceof ThingsboardKafkaClientError) { @@ -98,6 +98,10 @@ public class TbKafkaNode extends TbAbstractExternalNode { } } + protected KafkaProducer getKafkaProducer(Properties properties) { + return new KafkaProducer<>(properties); + } + @Override public void onMsg(TbContext ctx, TbMsg msg) { String topic = TbNodeUtils.processPattern(config.getTopicPattern(), msg); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java index 86c39336f4..6f3ba4b8c7 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java @@ -16,7 +16,7 @@ package org.thingsboard.rule.engine.kafka; import org.apache.kafka.clients.producer.Callback; -import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; @@ -24,6 +24,7 @@ import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.header.internals.RecordHeader; import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.kafka.common.utils.KafkaThread; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -31,6 +32,7 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.NullAndEmptySource; +import org.junit.jupiter.params.provider.ValueSource; import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; @@ -40,14 +42,16 @@ import org.thingsboard.common.util.ListeningExecutor; import org.thingsboard.rule.engine.TestDbCallbackExecutor; import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNodeConfiguration; +import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.rule.engine.api.util.TbNodeUtils; +import org.thingsboard.server.common.data.exception.ThingsboardKafkaClientError; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.msg.TbMsgType; +import org.thingsboard.server.common.data.msg.TbNodeConnectionType; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; -import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; import java.util.Map; import java.util.Properties; @@ -56,40 +60,54 @@ import java.util.concurrent.Callable; import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatNoException; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.mock; +import static org.mockito.BDDMockito.never; +import static org.mockito.BDDMockito.spy; import static org.mockito.BDDMockito.then; +import static org.mockito.BDDMockito.times; import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.BDDMockito.willReturn; import static org.mockito.BDDMockito.willThrow; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; @ExtendWith(MockitoExtension.class) public class TbKafkaNodeTest { private final DeviceId DEVICE_ID = new DeviceId(UUID.fromString("5f2eac08-bd1f-4635-a6c2-437369f996cf")); + private final RuleNodeId RULE_NODE_ID = new RuleNodeId(UUID.fromString("d46bb666-ecab-4d89-a28f-5abdca23ac29")); private final ListeningExecutor executor = new TestDbCallbackExecutor(); + private final long OFFSET = 1; + private final int PARTITION = 0; + + private final String TEST_TOPIC = "test-topic"; + private final String TEST_KEY = "test-key"; + private TbKafkaNode node; private TbKafkaNodeConfiguration config; @Mock private TbContext ctxMock; @Mock - private Producer producerMock; + private KafkaProducer producerMock; + @Mock + private KafkaThread ioThreadMock; @Mock private RecordMetadata recordMetadataMock; @BeforeEach - void setUp() { - node = new TbKafkaNode(); + public void setUp() { + node = spy(new TbKafkaNode()); config = new TbKafkaNodeConfiguration().defaultConfiguration(); + config.setTopicPattern(TEST_TOPIC); + config.setKeyPattern(TEST_KEY); } @Test public void verifyDefaultConfig() { + config = new TbKafkaNodeConfiguration().defaultConfiguration(); assertThat(config.getTopicPattern()).isEqualTo("my-topic"); assertThat(config.getKeyPattern()).isNull(); assertThat(config.getBootstrapServers()).isEqualTo("localhost:9092"); @@ -106,27 +124,27 @@ public class TbKafkaNodeTest { } @Test - public void givenAddMetadataKeyValuesAsKafkaHeadersIsTrueAndKafkaHeadersCharsetIsSet_whenInit_thenOk() { - config.setAddMetadataKeyValuesAsKafkaHeaders(true); - config.setKafkaHeadersCharset("UTF-16"); - - String ruleNodeIdStr = "0d35733c-7661-4797-819e-d9188974e3b2"; - String serviceIdStr = "test-service"; - - given(ctxMock.getSelfId()).willReturn(new RuleNodeId(UUID.fromString(ruleNodeIdStr))); - given(ctxMock.getServiceId()).willReturn(serviceIdStr); - - assertThatNoException().isThrownBy(() -> node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)))); + public void givenExceptionDuringKafkaInitialization_whenInit_thenDestroy() throws TbNodeException { + // GIVEN + given(ctxMock.getSelfId()).willReturn(RULE_NODE_ID); + ReflectionTestUtils.setField(producerMock, "ioThread", ioThreadMock); + willAnswer(invocationOnMock -> { + Thread.UncaughtExceptionHandler exceptionHandler = invocationOnMock.getArgument(0); + exceptionHandler.uncaughtException(ioThreadMock, new ThingsboardKafkaClientError("Error during init")); + return null; + }).given(ioThreadMock).setUncaughtExceptionHandler(any()); + willReturn(producerMock).given(node).getKafkaProducer(any()); - Boolean addMetadataKeyValuesAsKafkaHeaders = (Boolean) ReflectionTestUtils.getField(node, "addMetadataKeyValuesAsKafkaHeaders"); - Charset toBytesCharset = (Charset) ReflectionTestUtils.getField(node, "toBytesCharset"); + // WHEN + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); - assertThat(addMetadataKeyValuesAsKafkaHeaders).isTrue(); - assertThat(toBytesCharset).isEqualTo(StandardCharsets.UTF_16); + // THEN + then(producerMock).should().close(); + then(producerMock).shouldHaveNoMoreInteractions(); } @Test - public void verifyGetKafkaPropertiesMethod() { + public void verifyGetKafkaPropertiesMethod() throws TbNodeException { String sslKeyStoreCertificateChain = "cbdvch\\nfwrg\nvgwg\\n"; String sslKeyStoreKey = "nghmh\\nhmmnh\\\\ngreg\nvgwg\\n"; String sslTruststoreCertificates = "grthrt\fd\\nfwrg\nvgwg\\n"; @@ -136,15 +154,17 @@ public class TbKafkaNodeTest { "ssl.truststore.certificates", sslTruststoreCertificates, "ssl.protocol", "TLSv1.2" )); - ReflectionTestUtils.setField(node, "config", config); - String ruleNodeIdStr = "e646b885-8004-45b4-8bfb-78db21870e0f"; + ReflectionTestUtils.setField(producerMock, "ioThread", ioThreadMock); + given(ctxMock.getSelfId()).willReturn(RULE_NODE_ID); String serviceIdStr = "test-service"; - given(ctxMock.getSelfId()).willReturn(new RuleNodeId(UUID.fromString(ruleNodeIdStr))); given(ctxMock.getServiceId()).willReturn(serviceIdStr); + willReturn(producerMock).given(node).getKafkaProducer(any()); + + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); Properties expectedProperties = new Properties(); - expectedProperties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-tb-kafka-node-" + ruleNodeIdStr + "-" + serviceIdStr); + expectedProperties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-tb-kafka-node-" + RULE_NODE_ID.getId() + "-" + serviceIdStr); expectedProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers()); expectedProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, config.getValueSerializer()); expectedProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, config.getKeySerializer()); @@ -163,14 +183,18 @@ public class TbKafkaNodeTest { } @Test - public void givenInitErrorIsNotNull_whenOnMsg_thenTellFailure() { - init(); - String errorMsg = "Error during init!"; + public void givenInitErrorIsNotNull_whenOnMsg_thenTellFailure() throws TbNodeException { + // GIVEN + mockSuccessfulInit(); + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + String errorMsg = "Error during kafka initialization!"; ReflectionTestUtils.setField(node, "initError", new RuntimeException(errorMsg)); + // WHEN TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); node.onMsg(ctxMock, msg); + // THEN ArgumentCaptor actualError = ArgumentCaptor.forClass(Throwable.class); then(ctxMock).should().tellFailure(eq(msg), actualError.capture()); assertThat(actualError.getValue()) @@ -178,20 +202,24 @@ public class TbKafkaNodeTest { .hasMessage("Failed to initialize Kafka rule node producer: " + errorMsg); } - @Test - public void givenForceAckIsTrueAndExceptionWasThrown_whenOnMsg_thenTellFailure() { - init(); - ReflectionTestUtils.setField(node, "forceAck", true); - + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void givenForceAckAndExceptionWasThrown_whenOnMsg_thenTellFailure(boolean forceAck) throws TbNodeException { + // GIVEN + given(ctxMock.isExternalNodeForceAck()).willReturn(forceAck); + mockSuccessfulInit(); ListeningExecutor executorMock = mock(ListeningExecutor.class); given(ctxMock.getExternalCallExecutor()).willReturn(executorMock); String errorMsg = "Something went wrong!"; willThrow(new RuntimeException(errorMsg)).given(executorMock).executeAsync(any(Callable.class)); + // WHEN + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); node.onMsg(ctxMock, msg); - then(ctxMock).should().ack(msg); + // THEN + then(ctxMock).should(forceAck ? times(1) : never()).ack(msg); ArgumentCaptor actualMsg = ArgumentCaptor.forClass(TbMsg.class); ArgumentCaptor actualError = ArgumentCaptor.forClass(Throwable.class); then(ctxMock).should().tellFailure(actualMsg.capture(), actualError.capture()); @@ -201,26 +229,33 @@ public class TbKafkaNodeTest { @ParameterizedTest @MethodSource - public void givenTopicAndKeyPatternsAndAddMetadataKeyValuesAsKafkaHeadersIsFalse_whenOnMsg_thenTellSuccess - (String topicPattern, String keyPattern, TbMsgMetaData metaData, String data) { + public void givenForceAckIsTrueTopicAndKeyPatternsAndAddMetadataKeyValuesAsKafkaHeadersIsFalse_whenOnMsg_thenEnqueueForTellNext( + String topicPattern, String keyPattern, TbMsgMetaData metaData, String data + ) throws TbNodeException { + // GIVEN config.setTopicPattern(topicPattern); config.setKeyPattern(keyPattern); - init(); - TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, metaData, data); String topic = TbNodeUtils.processPattern(topicPattern, msg); String key = TbNodeUtils.processPattern(keyPattern, msg); - long offset = 1; - int partition = 0; - mockSuccessfulPublishingRequest(topic, offset, partition); + given(ctxMock.isExternalNodeForceAck()).willReturn(true); + mockSuccessfulInit(); + mockSuccessfulPublishingRequest(topic); + + // WHEN + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); node.onMsg(ctxMock, msg); + // THEN + then(ctxMock).should().ack(msg); verifyProducerRecord(topic, key, msg.getData()); - verifyOutboundMsg(offset, partition, topic, msg); + ArgumentCaptor actualMsg = ArgumentCaptor.forClass(TbMsg.class); + then(ctxMock).should().enqueueForTellNext(actualMsg.capture(), eq(TbNodeConnectionType.SUCCESS)); + verifyOutgoingSuccessMsg(topic, actualMsg.getValue(), msg); } - private static Stream givenTopicAndKeyPatternsAndAddMetadataKeyValuesAsKafkaHeadersIsFalse_whenOnMsg_thenTellSuccess() { + private static Stream givenForceAckIsTrueTopicAndKeyPatternsAndAddMetadataKeyValuesAsKafkaHeadersIsFalse_whenOnMsg_thenEnqueueForTellNext() { return Stream.of( Arguments.of("test-topic", "test-key", new TbMsgMetaData(), TbMsg.EMPTY_JSON_OBJECT), Arguments.of("${mdTopicPattern}", "${mdKeyPattern}", new TbMsgMetaData( @@ -233,79 +268,89 @@ public class TbKafkaNodeTest { ); } - @Test - public void givenForceAckIsFalseAndAddMetadataKeyValuesAsKafkaHeadersIsTrueAndToBytesCharsetIsSet_whenOnMsg_thenAckAndTellSuccess() { - String topic = "test-topic"; - String key = "test-key"; - config.setTopicPattern(topic); + @ParameterizedTest + @NullAndEmptySource + public void givenForceAckIsFalseAndKeyIsNullOrEmptyAndErrorOccursDuringPublishing_whenOnMsg_thenTellFailure(String key) throws TbNodeException { + // GIVEN config.setKeyPattern(key); - config.setAddMetadataKeyValuesAsKafkaHeaders(true); - config.setKafkaHeadersCharset("UTF-16"); - init(); - ReflectionTestUtils.setField(node, "forceAck", false); - ReflectionTestUtils.setField(node, "addMetadataKeyValuesAsKafkaHeaders", true); - ReflectionTestUtils.setField(node, "toBytesCharset", StandardCharsets.UTF_16); - - TbMsgMetaData metaData = new TbMsgMetaData(); - metaData.putValue("key", "value"); - TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, metaData, TbMsg.EMPTY_JSON_OBJECT); - - long offset = 1; - int partition = 0; - mockSuccessfulPublishingRequest(topic, offset, partition); + given(ctxMock.isExternalNodeForceAck()).willReturn(false); + mockSuccessfulInit(); + String errorMsg = "Something went wrong!"; + mockFailedPublishingRequest(new RuntimeException(errorMsg)); + // WHEN + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); node.onMsg(ctxMock, msg); + // THEN + verifyProducerRecord(TEST_TOPIC, null, msg.getData()); then(ctxMock).should(never()).ack(msg); - Headers expectedHeaders = new RecordHeaders(); - msg.getMetaData().values().forEach((k, v) -> expectedHeaders.add(new RecordHeader("tb_msg_md_" + k, v.getBytes(StandardCharsets.UTF_16)))); - verifyProducerRecord(topic, key, msg.getData(), expectedHeaders); - verifyOutboundMsg(offset, partition, topic, msg); + ArgumentCaptor actualMsg = ArgumentCaptor.forClass(TbMsg.class); + ArgumentCaptor actualError = ArgumentCaptor.forClass(Throwable.class); + then(ctxMock).should().tellFailure(actualMsg.capture(), actualError.capture()); + verifyOutgoingFailureMsg(errorMsg, actualMsg.getValue(), msg); } - @ParameterizedTest - @NullAndEmptySource - public void givenKeyIsNullOrEmptyAndErrorOccursDuringPublishing_whenOnMsg_thenTellFailure(String key) { - String topic = "test-topic"; - config.setTopicPattern(topic); - config.setKeyPattern(key); - config.setAddMetadataKeyValuesAsKafkaHeaders(false); - - TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); + @Test + public void givenForceAckIsTrueAndAddKafkaHeadersIsTrueAndToBytesCharsetIsNullAndErrorOccursDuringPublishing_whenOnMsg_thenEnqueueForTellFailure() throws TbNodeException { + // GIVEN + config.setAddMetadataKeyValuesAsKafkaHeaders(true); + config.setKafkaHeadersCharset(null); + given(ctxMock.isExternalNodeForceAck()).willReturn(true); + mockSuccessfulInit(); String errorMsg = "Something went wrong!"; + mockFailedPublishingRequest(new RuntimeException(errorMsg)); - given(ctxMock.getExternalCallExecutor()).willReturn(executor); - willAnswer(invocation -> { - Callback callback = invocation.getArgument(1); - callback.onCompletion(recordMetadataMock, new RuntimeException(errorMsg)); - return null; - }).given(producerMock).send(any(), any(Callback.class)); - - init(); + // WHEN + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT); node.onMsg(ctxMock, msg); - verifyProducerRecord(topic, null, msg.getData()); - + // THEN + then(ctxMock).should().ack(msg); + Headers expectedHeaders = new RecordHeaders(); + msg.getMetaData().values().forEach((k, v) -> expectedHeaders.add(new RecordHeader("tb_msg_md_" + k, v.getBytes(StandardCharsets.UTF_8)))); + verifyProducerRecord(TEST_TOPIC, TEST_KEY, msg.getData(), expectedHeaders); ArgumentCaptor actualMsg = ArgumentCaptor.forClass(TbMsg.class); ArgumentCaptor actualError = ArgumentCaptor.forClass(Throwable.class); - then(ctxMock).should().tellFailure(actualMsg.capture(), actualError.capture()); + then(ctxMock).should().enqueueForTellFailure(actualMsg.capture(), actualError.capture()); + verifyOutgoingFailureMsg(errorMsg, actualMsg.getValue(), msg); + } + + @Test + public void givenForceAckIsFalseAndAddMetadataKeyValuesAsKafkaHeadersIsTrueAndToBytesCharsetIsSet_whenOnMsg_thenTellSuccess() throws TbNodeException { + // GIVEN + config.setAddMetadataKeyValuesAsKafkaHeaders(true); + config.setKafkaHeadersCharset("UTF-16"); + + given(ctxMock.isExternalNodeForceAck()).willReturn(false); + mockSuccessfulInit(); + mockSuccessfulPublishingRequest(TEST_TOPIC); + + // WHEN + node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); TbMsgMetaData metaData = new TbMsgMetaData(); - metaData.putValue("error", RuntimeException.class + ": " + errorMsg); - TbMsg expectedMsg = TbMsg.transformMsgMetadata(msg, metaData); - assertThat(actualMsg.getValue()) - .usingRecursiveComparison() - .ignoringFields("ctx") - .isEqualTo(expectedMsg); + metaData.putValue("key", "value"); + TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, metaData, TbMsg.EMPTY_JSON_OBJECT); + node.onMsg(ctxMock, msg); + + // THEN + then(ctxMock).should(never()).ack(msg); + Headers expectedHeaders = new RecordHeaders(); + msg.getMetaData().values().forEach((k, v) -> expectedHeaders.add(new RecordHeader("tb_msg_md_" + k, v.getBytes(StandardCharsets.UTF_16)))); + verifyProducerRecord(TEST_TOPIC, TEST_KEY, msg.getData(), expectedHeaders); + ArgumentCaptor actualMsg = ArgumentCaptor.forClass(TbMsg.class); + then(ctxMock).should().tellSuccess(actualMsg.capture()); + verifyOutgoingSuccessMsg(TEST_TOPIC, actualMsg.getValue(), msg); } @Test public void givenProducerIsNotNull_whenDestroy_thenShouldClose() { ReflectionTestUtils.setField(node, "producer", producerMock); - node.destroy(); - then(producerMock).should().close(); } @@ -315,22 +360,31 @@ public class TbKafkaNodeTest { then(producerMock).shouldHaveNoInteractions(); } - private void mockSuccessfulPublishingRequest(String topic, long offset, int partition) { + private void mockSuccessfulInit() { + ReflectionTestUtils.setField(producerMock, "ioThread", ioThreadMock); + willReturn(mock(Properties.class)).given(node).getKafkaProperties(ctxMock); + willReturn(producerMock).given(node).getKafkaProducer(any()); + } + + private void mockSuccessfulPublishingRequest(String topic) { given(ctxMock.getExternalCallExecutor()).willReturn(executor); willAnswer(invocation -> { Callback callback = invocation.getArgument(1); callback.onCompletion(recordMetadataMock, null); return null; }).given(producerMock).send(any(), any(Callback.class)); - given(recordMetadataMock.offset()).willReturn(offset); - given(recordMetadataMock.partition()).willReturn(partition); + given(recordMetadataMock.offset()).willReturn(OFFSET); + given(recordMetadataMock.partition()).willReturn(PARTITION); given(recordMetadataMock.topic()).willReturn(topic); } - private void init() { - ReflectionTestUtils.setField(node, "config", config); - ReflectionTestUtils.setField(node, "producer", producerMock); - ReflectionTestUtils.setField(node, "addMetadataKeyValuesAsKafkaHeaders", false); + private void mockFailedPublishingRequest(Exception exception) { + given(ctxMock.getExternalCallExecutor()).willReturn(executor); + willAnswer(invocation -> { + Callback callback = invocation.getArgument(1); + callback.onCompletion(recordMetadataMock, exception); + return null; + }).given(producerMock).send(any(), any(Callback.class)); } private void verifyProducerRecord(String expectedTopic, String expectedKey, String expectedValue) { @@ -349,17 +403,23 @@ public class TbKafkaNodeTest { } } - private void verifyOutboundMsg(long expectedOffset, long expectedPartition, String expectedTopic, TbMsg originalMsg) { - ArgumentCaptor actualMsg = ArgumentCaptor.forClass(TbMsg.class); - then(ctxMock).should().tellSuccess(actualMsg.capture()); + private void verifyOutgoingSuccessMsg(String expectedTopic, TbMsg actualMsg, TbMsg originalMsg) { TbMsgMetaData metaData = originalMsg.getMetaData().copy(); - metaData.putValue("offset", String.valueOf(expectedOffset)); - metaData.putValue("partition", String.valueOf(expectedPartition)); + metaData.putValue("offset", String.valueOf(OFFSET)); + metaData.putValue("partition", String.valueOf(PARTITION)); metaData.putValue("topic", expectedTopic); TbMsg expectedMsg = TbMsg.transformMsgMetadata(originalMsg, metaData); - assertThat(actualMsg.getValue()) + assertThat(actualMsg) .usingRecursiveComparison() .ignoringFields("ctx") .isEqualTo(expectedMsg); } + + private void verifyOutgoingFailureMsg(String errorMsg, TbMsg actualMsg, TbMsg originalMsg) { + TbMsgMetaData metaData = originalMsg.getMetaData(); + metaData.putValue("error", RuntimeException.class + ": " + errorMsg); + TbMsg expectedMsg = TbMsg.transformMsgMetadata(originalMsg, metaData); + assertThat(actualMsg).usingRecursiveComparison().ignoringFields("ctx").isEqualTo(expectedMsg); + } + } From 62699900bb125cb86fe0c11aa9e7b855c0941cfb Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Wed, 7 Aug 2024 10:08:48 +0300 Subject: [PATCH 3/4] moved properties creation to init() method --- .../rule/engine/kafka/TbKafkaNode.java | 47 +++++++++---------- .../rule/engine/kafka/TbKafkaNodeTest.java | 19 ++++---- 2 files changed, 30 insertions(+), 36 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java index 1b77e5c780..89b6e1c1d9 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java @@ -81,7 +81,26 @@ public class TbKafkaNode extends TbAbstractExternalNode { super.init(ctx); this.config = TbNodeUtils.convert(configuration, TbKafkaNodeConfiguration.class); this.initError = null; - Properties properties = getKafkaProperties(ctx); + Properties properties = new Properties(); + properties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-tb-kafka-node-" + ctx.getSelfId().getId().toString() + "-" + ctx.getServiceId()); + properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers()); + properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, config.getValueSerializer()); + properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, config.getKeySerializer()); + properties.put(ProducerConfig.ACKS_CONFIG, config.getAcks()); + properties.put(ProducerConfig.RETRIES_CONFIG, config.getRetries()); + properties.put(ProducerConfig.BATCH_SIZE_CONFIG, config.getBatchSize()); + properties.put(ProducerConfig.LINGER_MS_CONFIG, config.getLinger()); + properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, config.getBufferMemory()); + if (config.getOtherProperties() != null) { + config.getOtherProperties().forEach((k, v) -> { + if (SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG.equals(k) + || SslConfigs.SSL_KEYSTORE_KEY_CONFIG.equals(k) + || SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG.equals(k)) { + v = v.replace("\\n", "\n"); + } + properties.put(k, v); + }); + } addMetadataKeyValuesAsKafkaHeaders = BooleanUtils.toBooleanDefaultIfNull(config.isAddMetadataKeyValuesAsKafkaHeaders(), false); toBytesCharset = config.getKafkaHeadersCharset() != null ? Charset.forName(config.getKafkaHeadersCharset()) : StandardCharsets.UTF_8; try { @@ -98,7 +117,7 @@ public class TbKafkaNode extends TbAbstractExternalNode { } } - protected KafkaProducer getKafkaProducer(Properties properties) { + KafkaProducer getKafkaProducer(Properties properties) { return new KafkaProducer<>(properties); } @@ -145,30 +164,6 @@ public class TbKafkaNode extends TbAbstractExternalNode { } } - protected Properties getKafkaProperties(TbContext ctx) { - Properties properties = new Properties(); - properties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-tb-kafka-node-" + ctx.getSelfId().getId().toString() + "-" + ctx.getServiceId()); - properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers()); - properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, config.getValueSerializer()); - properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, config.getKeySerializer()); - properties.put(ProducerConfig.ACKS_CONFIG, config.getAcks()); - properties.put(ProducerConfig.RETRIES_CONFIG, config.getRetries()); - properties.put(ProducerConfig.BATCH_SIZE_CONFIG, config.getBatchSize()); - properties.put(ProducerConfig.LINGER_MS_CONFIG, config.getLinger()); - properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, config.getBufferMemory()); - if (config.getOtherProperties() != null) { - config.getOtherProperties().forEach((k, v) -> { - if (SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG.equals(k) - || SslConfigs.SSL_KEYSTORE_KEY_CONFIG.equals(k) - || SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG.equals(k)) { - v = v.replace("\\n", "\n"); - } - properties.put(k, v); - }); - } - return properties; - } - @Override public void destroy() { if (this.producer != null) { diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java index 6f3ba4b8c7..e89ac2bd10 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java @@ -82,6 +82,7 @@ public class TbKafkaNodeTest { private final long OFFSET = 1; private final int PARTITION = 0; + private final String SERVICE_ID_STR = "test-service-id"; private final String TEST_TOPIC = "test-topic"; private final String TEST_KEY = "test-key"; @@ -144,7 +145,7 @@ public class TbKafkaNodeTest { } @Test - public void verifyGetKafkaPropertiesMethod() throws TbNodeException { + public void verifyKafkaProperties() throws TbNodeException { String sslKeyStoreCertificateChain = "cbdvch\\nfwrg\nvgwg\\n"; String sslKeyStoreKey = "nghmh\\nhmmnh\\\\ngreg\nvgwg\\n"; String sslTruststoreCertificates = "grthrt\fd\\nfwrg\nvgwg\\n"; @@ -155,16 +156,12 @@ public class TbKafkaNodeTest { "ssl.protocol", "TLSv1.2" )); - ReflectionTestUtils.setField(producerMock, "ioThread", ioThreadMock); - given(ctxMock.getSelfId()).willReturn(RULE_NODE_ID); - String serviceIdStr = "test-service"; - given(ctxMock.getServiceId()).willReturn(serviceIdStr); - willReturn(producerMock).given(node).getKafkaProducer(any()); + mockSuccessfulInit(); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); Properties expectedProperties = new Properties(); - expectedProperties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-tb-kafka-node-" + RULE_NODE_ID.getId() + "-" + serviceIdStr); + expectedProperties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-tb-kafka-node-" + RULE_NODE_ID.getId() + "-" + SERVICE_ID_STR); expectedProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers()); expectedProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, config.getValueSerializer()); expectedProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, config.getKeySerializer()); @@ -178,8 +175,9 @@ public class TbKafkaNodeTest { expectedProperties.put("ssl.truststore.certificates", sslTruststoreCertificates.replace("\\n", "\n")); expectedProperties.put("ssl.protocol", "TLSv1.2"); - Properties actualsProperties = node.getKafkaProperties(ctxMock); - assertThat(actualsProperties).isEqualTo(expectedProperties); + ArgumentCaptor properties = ArgumentCaptor.forClass(Properties.class); + then(node).should().getKafkaProducer(properties.capture()); + assertThat(properties.getValue()).isEqualTo(expectedProperties); } @Test @@ -361,8 +359,9 @@ public class TbKafkaNodeTest { } private void mockSuccessfulInit() { + given(ctxMock.getSelfId()).willReturn(RULE_NODE_ID); + given(ctxMock.getServiceId()).willReturn(SERVICE_ID_STR); ReflectionTestUtils.setField(producerMock, "ioThread", ioThreadMock); - willReturn(mock(Properties.class)).given(node).getKafkaProperties(ctxMock); willReturn(producerMock).given(node).getKafkaProducer(any()); } From a5ae204b585fd2b0abe89d2556facf6a49f14d71 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Wed, 7 Aug 2024 13:17:24 +0300 Subject: [PATCH 4/4] changed initialization for test --- .../org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java index e89ac2bd10..add12dcec2 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java @@ -181,12 +181,11 @@ public class TbKafkaNodeTest { } @Test - public void givenInitErrorIsNotNull_whenOnMsg_thenTellFailure() throws TbNodeException { + public void givenInitErrorIsNotNull_whenOnMsg_thenTellFailure() { // GIVEN - mockSuccessfulInit(); - node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); String errorMsg = "Error during kafka initialization!"; - ReflectionTestUtils.setField(node, "initError", new RuntimeException(errorMsg)); + 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);