Browse Source

Fix topics creation for isolated tenants

pull/13799/head
VIacheslavKlimov 12 months ago
parent
commit
01e72f4e30
  1. 4
      application/src/main/java/org/thingsboard/server/service/entitiy/queue/DefaultTbQueueService.java
  2. 5
      common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueAdmin.java
  3. 2
      common/queue/src/main/java/org/thingsboard/server/queue/RuleEngineTbQueueAdminFactory.java
  4. 10
      common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaAdmin.java
  5. 2
      common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryTbTransportQueueFactory.java

4
application/src/main/java/org/thingsboard/server/service/entitiy/queue/DefaultTbQueueService.java

@ -176,8 +176,8 @@ public class DefaultTbQueueService extends AbstractTbEntityService implements Tb
for (int i = oldPartitions; i < newPartitions; i++) {
tbQueueAdmin.createTopicIfNotExists(
new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName(),
queue.getCustomProperties()
);
queue.getCustomProperties(),
true); // forcing topic creation because the topic may still be cached on some nodes
}
}

5
common/cluster-api/src/main/java/org/thingsboard/server/queue/TbQueueAdmin.java

@ -18,12 +18,13 @@ package org.thingsboard.server.queue;
public interface TbQueueAdmin {
default void createTopicIfNotExists(String topic) {
createTopicIfNotExists(topic, null);
createTopicIfNotExists(topic, null, false);
}
void createTopicIfNotExists(String topic, String properties);
void createTopicIfNotExists(String topic, String properties, boolean force);
void destroy();
void deleteTopic(String topic);
}

2
common/queue/src/main/java/org/thingsboard/server/queue/RuleEngineTbQueueAdminFactory.java

@ -43,7 +43,7 @@ public class RuleEngineTbQueueAdminFactory {
return new TbQueueAdmin() {
@Override
public void createTopicIfNotExists(String topic, String properties) {
public void createTopicIfNotExists(String topic, String properties, boolean force) {
}
@Override

10
common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaAdmin.java

@ -70,10 +70,12 @@ public class TbKafkaAdmin implements TbQueueAdmin, TbEdgeQueueAdmin {
}
@Override
public void createTopicIfNotExists(String topic, String properties) {
Set<String> topics = getTopics();
if (topics.contains(topic)) {
return;
public void createTopicIfNotExists(String topic, String properties, boolean force) {
if (!force) {
Set<String> topics = getTopics();
if (topics.contains(topic)) {
return;
}
}
try {
Map<String, String> configs = PropertyUtils.getProps(topicConfigs, properties);

2
common/queue/src/main/java/org/thingsboard/server/queue/provider/InMemoryTbTransportQueueFactory.java

@ -78,7 +78,7 @@ public class InMemoryTbTransportQueueFactory implements TbTransportQueueFactory
templateBuilder.queueAdmin(new TbQueueAdmin() {
@Override
public void createTopicIfNotExists(String topic, String properties) {}
public void createTopicIfNotExists(String topic, String properties, boolean force) {}
@Override
public void destroy() {}

Loading…
Cancel
Save