tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge pull request #11037 from irynamatveieva/improvements/kafka-node 

Kafka node: added tests
pull/11393/head
Viacheslav Klimov 2 years ago
committed by GitHub
parent
a7e81e7ca7 a5ae204b58
commit
2b39bb8cd8
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
2 changed files with 428 additions and 1 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 6
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java
  2. 423
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java

6
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java
View File

@ -104,7 +104,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) {
@ -117,6 +117,10 @@ public class TbKafkaNode extends TbAbstractExternalNode {
}
}
KafkaProducer<String, String> getKafkaProducer(Properties properties) {
return new KafkaProducer<>(properties);
}
@Override
public void onMsg(TbContext ctx, TbMsg msg) {
String topic = TbNodeUtils.processPattern(config.getTopicPattern(), msg);

423
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/kafka/TbKafkaNodeTest.java
View File

@ -0,0 +1,423 @@
/**
* 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.KafkaProducer;
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.apache.kafka.common.utils.KafkaThread;
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.junit.jupiter.params.provider.ValueSource;
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.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.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.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;
@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 SERVICE_ID_STR = "test-service-id";
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 KafkaProducer<String, String> producerMock;
@Mock
private KafkaThread ioThreadMock;
@Mock
private RecordMetadata recordMetadataMock;
@BeforeEach
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");
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 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());
// WHEN
node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
// THEN
then(producerMock).should().close();
then(producerMock).shouldHaveNoMoreInteractions();
}
@Test
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";
config.setOtherProperties(Map.of(
"ssl.keystore.certificate.chain", sslKeyStoreCertificateChain,
"ssl.keystore.key", sslKeyStoreKey,
"ssl.truststore.certificates", sslTruststoreCertificates,
"ssl.protocol", "TLSv1.2"
));
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() + "-" + 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());
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");
ArgumentCaptor<Properties> properties = ArgumentCaptor.forClass(Properties.class);
then(node).should().getKafkaProducer(properties.capture());
assertThat(properties.getValue()).isEqualTo(expectedProperties);
}
@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);
}
@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
then(ctxMock).should(forceAck ? times(1) : never()).ack(msg);
ArgumentCaptor<TbMsg> actualMsg = ArgumentCaptor.forClass(TbMsg.class);
ArgumentCaptor<Throwable> 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 givenForceAckIsTrueTopicAndKeyPatternsAndAddMetadataKeyValuesAsKafkaHeadersIsFalse_whenOnMsg_thenEnqueueForTellNext(
String topicPattern, String keyPattern, TbMsgMetaData metaData, String data
) throws TbNodeException {
// GIVEN
config.setTopicPattern(topicPattern);
config.setKeyPattern(keyPattern);
TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, DEVICE_ID, metaData, data);
String topic = TbNodeUtils.processPattern(topicPattern, msg);
String key = TbNodeUtils.processPattern(keyPattern, msg);
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());
ArgumentCaptor<TbMsg> actualMsg = ArgumentCaptor.forClass(TbMsg.class);
then(ctxMock).should().enqueueForTellNext(actualMsg.capture(), eq(TbNodeConnectionType.SUCCESS));
verifyOutgoingSuccessMsg(topic, actualMsg.getValue(), msg);
}
private static Stream<Arguments> givenForceAckIsTrueTopicAndKeyPatternsAndAddMetadataKeyValuesAsKafkaHeadersIsFalse_whenOnMsg_thenEnqueueForTellNext() {
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\"}")
);
}
@ParameterizedTest
@NullAndEmptySource
public void givenForceAckIsFalseAndKeyIsNullOrEmptyAndErrorOccursDuringPublishing_whenOnMsg_thenTellFailure(String key) throws TbNodeException {
// GIVEN
config.setKeyPattern(key);
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);
ArgumentCaptor<TbMsg> actualMsg = ArgumentCaptor.forClass(TbMsg.class);
ArgumentCaptor<Throwable> actualError = ArgumentCaptor.forClass(Throwable.class);
then(ctxMock).should().tellFailure(actualMsg.capture(), actualError.capture());
verifyOutgoingFailureMsg(errorMsg, actualMsg.getValue(), msg);
}
@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));
// 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
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<TbMsg> actualMsg = ArgumentCaptor.forClass(TbMsg.class);
ArgumentCaptor<Throwable> actualError = ArgumentCaptor.forClass(Throwable.class);
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("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<TbMsg> 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();
}
@Test
public void givenProducerIsNull_whenDestroy_thenDoNothing() {
node.destroy();
then(producerMock).shouldHaveNoInteractions();
}
private void mockSuccessfulInit() {
given(ctxMock.getSelfId()).willReturn(RULE_NODE_ID);
given(ctxMock.getServiceId()).willReturn(SERVICE_ID_STR);
ReflectionTestUtils.setField(producerMock, "ioThread", ioThreadMock);
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.topic()).willReturn(topic);
}
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) {
verifyProducerRecord(expectedTopic, expectedKey, expectedValue, null);
}
private void verifyProducerRecord(String expectedTopic, String expectedKey, String expectedValue, Headers expectedHeaders) {
ArgumentCaptor<ProducerRecord<String, String>> actualRecordCaptor = ArgumentCaptor.forClass(ProducerRecord.class);
then(producerMock).should().send(actualRecordCaptor.capture(), any());
ProducerRecord<String, String> 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 verifyOutgoingSuccessMsg(String expectedTopic, TbMsg actualMsg, TbMsg originalMsg) {
TbMsgMetaData metaData = originalMsg.getMetaData().copy();
metaData.putValue("offset", String.valueOf(OFFSET));
metaData.putValue("partition", String.valueOf(PARTITION));
metaData.putValue("topic", expectedTopic);
TbMsg expectedMsg = TbMsg.transformMsgMetadata(originalMsg, metaData);
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);
}
}
Write Preview
Loading…
Cancel
Save