Skip to content

Commit 1509707

Browse files
dmytro-landiakdmytro-landiak
authored
Merge pull request #377 from thingsboard/fix/state-topic-init-dummy-persist-retry
fix(init): retry the startup marker write on fresh Kafka topics
2 parents 9f82278 + 946e586 commit 1509707

7 files changed

Lines changed: 202 additions & 6 deletions

File tree

‎application/src/main/java/org/thingsboard/mqtt/broker/service/mqtt/client/blocked/consumer/BlockedClientConsumerServiceImpl.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@
3838
import org.thingsboard.mqtt.broker.service.mqtt.client.blocked.data.BlockedClient;
3939
import org.thingsboard.mqtt.broker.service.mqtt.client.blocked.data.ClientIdBlockedClient;
4040
import org.thingsboard.mqtt.broker.service.mqtt.client.blocked.producer.BlockedClientProducerService;
41+
import org.thingsboard.mqtt.broker.util.InitLoadUtil;
4142

4243
import java.util.Collections;
4344
import java.util.HashMap;
@@ -171,7 +172,8 @@ private BlockedClient convertToBlockedClient(TbProtoQueueMsg<BlockedClientProto>
171172
private String persistDummyBlockedClient() throws QueuePersistenceException {
172173
BlockedClient blockedClient = new ClientIdBlockedClient(UUIDUtil.randomUuid());
173174
String dummyBlockedClientKey = blockedClient.getKey();
174-
producerService.persistDummyBlockedClient(dummyBlockedClientKey, ProtoConverter.convertToBlockedClientProto(blockedClient));
175+
BlockedClientProto blockedClientProto = ProtoConverter.convertToBlockedClientProto(blockedClient);
176+
InitLoadUtil.persistMarkerWithRetry("blockedClient", () -> producerService.persistDummyBlockedClient(dummyBlockedClientKey, blockedClientProto));
175177
return dummyBlockedClientKey;
176178
}
177179

‎application/src/main/java/org/thingsboard/mqtt/broker/service/mqtt/client/session/ClientSessionConsumerImpl.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
import org.thingsboard.mqtt.broker.queue.common.TbProtoQueueMsg;
3535
import org.thingsboard.mqtt.broker.queue.constants.QueueConstants;
3636
import org.thingsboard.mqtt.broker.queue.provider.ClientSessionQueueFactory;
37+
import org.thingsboard.mqtt.broker.util.InitLoadUtil;
3738

3839
import java.util.Collections;
3940
import java.util.HashMap;
@@ -169,7 +170,8 @@ public void listen(ClientSessionChangesCallback callback) {
169170
private String persistDummySession() throws QueuePersistenceException {
170171
String dummyClientId = UUID.randomUUID().toString();
171172
ClientSessionInfo dummyClientSessionInfo = getClientSessionInfo(dummyClientId, serviceInfoProvider.getServiceId(), false);
172-
persistenceService.persistClientSessionInfoSync(dummyClientId, ProtoConverter.convertToClientSessionInfoProto(dummyClientSessionInfo));
173+
ClientSessionInfoProto dummyClientSessionInfoProto = ProtoConverter.convertToClientSessionInfoProto(dummyClientSessionInfo);
174+
InitLoadUtil.persistMarkerWithRetry("clientSession", () -> persistenceService.persistClientSessionInfoSync(dummyClientId, dummyClientSessionInfoProto));
173175
return dummyClientId;
174176
}
175177

‎application/src/main/java/org/thingsboard/mqtt/broker/service/mqtt/retain/RetainedMsgConsumerImpl.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import org.thingsboard.mqtt.broker.queue.provider.RetainedMsgQueueFactory;
3737
import org.thingsboard.mqtt.broker.service.stats.RetainedMsgConsumerStats;
3838
import org.thingsboard.mqtt.broker.service.stats.StatsManager;
39+
import org.thingsboard.mqtt.broker.util.InitLoadUtil;
3940
import org.thingsboard.mqtt.broker.util.MqttPropertiesUtil;
4041

4142
import java.util.Collections;
@@ -196,7 +197,8 @@ private static RetainedMsg convertToRetainedMsg(TbProtoQueueMsg<RetainedMsgProto
196197
private String persistDummyRetainedMsg() throws QueuePersistenceException {
197198
String dummyTopic = DUMMY_TOPIC_PREFIX + RandomStringUtils.randomAlphanumeric(8);
198199
RetainedMsg retainedMsg = new RetainedMsg(dummyTopic, BrokerConstants.DUMMY_PAYLOAD, 0);
199-
persistenceService.persistRetainedMsgSync(dummyTopic, ProtoConverter.convertToRetainedMsgProto(retainedMsg));
200+
RetainedMsgProto retainedMsgProto = ProtoConverter.convertToRetainedMsgProto(retainedMsg);
201+
InitLoadUtil.persistMarkerWithRetry("retainedMsg", () -> persistenceService.persistRetainedMsgSync(dummyTopic, retainedMsgProto));
200202
return dummyTopic;
201203
}
202204

‎application/src/main/java/org/thingsboard/mqtt/broker/service/subscription/ClientSubscriptionConsumerImpl.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import org.thingsboard.mqtt.broker.service.stats.StatsManager;
3838
import org.thingsboard.mqtt.broker.service.subscription.data.SourcedSubscriptions;
3939
import org.thingsboard.mqtt.broker.service.subscription.data.SubscriptionsSourceKey;
40+
import org.thingsboard.mqtt.broker.util.InitLoadUtil;
4041

4142
import java.util.Collections;
4243
import java.util.HashMap;
@@ -193,7 +194,8 @@ void processPack(List<TbProtoQueueMsg<ClientSubscriptionsProto>> messages, Clien
193194

194195
private String persistDummyClientSubscriptions() throws QueuePersistenceException {
195196
String dummyClientId = BrokerConstants.SYSTEM_DUMMY_CLIENT_ID_PREFIX + UUID.randomUUID();
196-
persistenceService.persistClientSubscriptionsSync(dummyClientId, Collections.singleton(new ClientTopicSubscription(BrokerConstants.SYSTEM_DUMMY_TOPIC_FILTER, 0)));
197+
Set<TopicSubscription> dummySubscriptions = Collections.singleton(new ClientTopicSubscription(BrokerConstants.SYSTEM_DUMMY_TOPIC_FILTER, 0));
198+
InitLoadUtil.persistMarkerWithRetry("clientSubscriptions", () -> persistenceService.persistClientSubscriptionsSync(dummyClientId, dummySubscriptions));
197199
return dummyClientId;
198200
}
199201

‎application/src/main/java/org/thingsboard/mqtt/broker/util/InitLoadUtil.java‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
/**
2+
* Copyright © 2016-2026 The Thingsboard Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package org.thingsboard.mqtt.broker.util;
17+
18+
import lombok.extern.slf4j.Slf4j;
19+
import org.thingsboard.mqtt.broker.exception.QueuePersistenceException;
20+
21+
@Slf4j
22+
public final class InitLoadUtil {
23+
24+
static final int MARKER_PERSIST_MAX_ATTEMPTS = 10;
25+
static final long MARKER_PERSIST_RETRY_PAUSE_MS = 1000;
26+
27+
private InitLoadUtil() {
28+
}
29+
30+
@FunctionalInterface
31+
public interface QueuePersistenceAction {
32+
void run() throws QueuePersistenceException;
33+
}
34+
35+
/**
36+
* Persists the marker record that a compacted-topic initLoad reads back to detect it has caught up.
37+
* <p>
38+
* On a fresh Kafka cluster the topic is created synchronously right before this first write, and the partition
39+
* leader may not have applied the new topic metadata yet — the produce fails with UNKNOWN_TOPIC_OR_PARTITION
40+
* once the producer's few retries are spent. Retrying here keeps that transient window from failing startup.
41+
*/
42+
public static void persistMarkerWithRetry(String name, QueuePersistenceAction action) throws QueuePersistenceException {
43+
persistMarkerWithRetry(name, action, MARKER_PERSIST_MAX_ATTEMPTS, MARKER_PERSIST_RETRY_PAUSE_MS);
44+
}
45+
46+
static void persistMarkerWithRetry(String name, QueuePersistenceAction action,
47+
int maxAttempts, long pauseMs) throws QueuePersistenceException {
48+
for (int attempt = 1; ; attempt++) {
49+
try {
50+
action.run();
51+
return;
52+
} catch (QueuePersistenceException e) {
53+
if (attempt >= maxAttempts) {
54+
throw e;
55+
}
56+
log.warn("[{}] Failed to persist initLoad marker (attempt {}/{}), retrying in {} ms. Reason - {}",
57+
name, attempt, maxAttempts, pauseMs, e.getMessage());
58+
sleep(pauseMs, e);
59+
}
60+
}
61+
}
62+
63+
private static void sleep(long pauseMs, QueuePersistenceException cause) throws QueuePersistenceException {
64+
if (pauseMs <= 0) {
65+
return;
66+
}
67+
try {
68+
Thread.sleep(pauseMs);
69+
} catch (InterruptedException ie) {
70+
Thread.currentThread().interrupt();
71+
throw cause;
72+
}
73+
}
74+
}

‎application/src/test/java/org/thingsboard/mqtt/broker/service/mqtt/retain/RetainedMsgConsumerImplTest.java‎

Lines changed: 51 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,20 +19,31 @@
1919
import org.junit.Test;
2020
import org.junit.runner.RunWith;
2121
import org.mockito.junit.MockitoJUnitRunner;
22+
import org.thingsboard.mqtt.broker.adaptor.ProtoConverter;
2223
import org.thingsboard.mqtt.broker.common.data.BrokerConstants;
24+
import org.thingsboard.mqtt.broker.exception.QueuePersistenceException;
2325
import org.thingsboard.mqtt.broker.gen.queue.RetainedMsgProto;
2426
import org.thingsboard.mqtt.broker.queue.TbQueueAdmin;
27+
import org.thingsboard.mqtt.broker.queue.TbQueueControlledOffsetConsumer;
2528
import org.thingsboard.mqtt.broker.queue.cluster.ServiceInfoProvider;
2629
import org.thingsboard.mqtt.broker.queue.common.DefaultTbQueueMsgHeaders;
2730
import org.thingsboard.mqtt.broker.queue.common.TbProtoQueueMsg;
2831
import org.thingsboard.mqtt.broker.queue.provider.RetainedMsgQueueFactory;
2932
import org.thingsboard.mqtt.broker.service.stats.RetainedMsgConsumerStats;
3033
import org.thingsboard.mqtt.broker.service.stats.StatsManager;
3134

35+
import java.util.ArrayList;
3236
import java.util.List;
37+
import java.util.Map;
3338

39+
import static org.junit.Assert.assertEquals;
3440
import static org.junit.Assert.assertThrows;
41+
import static org.junit.Assert.assertTrue;
42+
import static org.mockito.ArgumentMatchers.any;
3543
import static org.mockito.ArgumentMatchers.anyInt;
44+
import static org.mockito.ArgumentMatchers.anyLong;
45+
import static org.mockito.ArgumentMatchers.anyString;
46+
import static org.mockito.Mockito.doAnswer;
3647
import static org.mockito.Mockito.mock;
3748
import static org.mockito.Mockito.never;
3849
import static org.mockito.Mockito.verify;
@@ -43,6 +54,8 @@ public class RetainedMsgConsumerImplTest {
4354

4455
private RetainedMsgConsumerStats stats;
4556
private RetainedMsgChangesCallback callback;
57+
private RetainedMsgQueueFactory queueFactory;
58+
private RetainedMsgPersistenceService persistenceService;
4659
private RetainedMsgConsumerImpl consumer;
4760

4861
@Before
@@ -51,10 +64,12 @@ public void setUp() {
5164
stats = mock(RetainedMsgConsumerStats.class);
5265
when(statsManager.getRetainedMsgConsumerStats()).thenReturn(stats);
5366
callback = mock(RetainedMsgChangesCallback.class);
67+
queueFactory = mock(RetainedMsgQueueFactory.class);
68+
persistenceService = mock(RetainedMsgPersistenceService.class);
5469
consumer = new RetainedMsgConsumerImpl(
55-
mock(RetainedMsgQueueFactory.class),
70+
queueFactory,
5671
mock(ServiceInfoProvider.class),
57-
mock(RetainedMsgPersistenceService.class),
72+
persistenceService,
5873
mock(TbQueueAdmin.class),
5974
statsManager);
6075
}
@@ -86,6 +101,40 @@ public void givenMidPackFailure_whenProcessPack_thenNoCounterAdvances() {
86101
verify(stats, never()).log(anyInt(), anyInt());
87102
}
88103

104+
@Test
105+
@SuppressWarnings("unchecked")
106+
public void givenFirstMarkerWriteFailsOnFreshTopic_whenInitLoad_thenRetriesAndCompletes() throws Exception {
107+
// On a fresh Kafka cluster the first write to the just-created topic can fail while the partition leader has not
108+
// applied the new topic metadata yet ("This server does not host this topic-partition"). initLoad must retry the
109+
// marker write instead of failing broker startup.
110+
TbQueueControlledOffsetConsumer<TbProtoQueueMsg<RetainedMsgProto>> queueConsumer = mock(TbQueueControlledOffsetConsumer.class);
111+
when(queueFactory.createConsumer(any(), any())).thenReturn(queueConsumer);
112+
consumer.init();
113+
114+
List<String> persistedTopics = new ArrayList<>();
115+
doAnswer(invocation -> {
116+
persistedTopics.add(invocation.getArgument(0));
117+
if (persistedTopics.size() == 1) {
118+
throw new QueuePersistenceException("Failed to update retained msg");
119+
}
120+
return null;
121+
}).when(persistenceService).persistRetainedMsgSync(anyString(), any());
122+
when(queueConsumer.poll(anyLong())).thenAnswer(invocation -> List.of(markerMsg(persistedTopics.get(0))));
123+
124+
Map<String, RetainedMsg> result = consumer.initLoad();
125+
126+
assertTrue(result.isEmpty());
127+
// failed marker write + retried marker write (same key, so the consumer still recognizes it) + marker clear
128+
assertEquals(3, persistedTopics.size());
129+
assertEquals(persistedTopics.get(0), persistedTopics.get(1));
130+
assertEquals(persistedTopics.get(0), persistedTopics.get(2));
131+
}
132+
133+
private static TbProtoQueueMsg<RetainedMsgProto> markerMsg(String topic) {
134+
RetainedMsgProto proto = ProtoConverter.convertToRetainedMsgProto(new RetainedMsg(topic, BrokerConstants.DUMMY_PAYLOAD, 0));
135+
return new TbProtoQueueMsg<>(topic, proto, new DefaultTbQueueMsgHeaders());
136+
}
137+
89138
private static TbProtoQueueMsg<RetainedMsgProto> clearedMsg() {
90139
// Default RetainedMsgProto has an empty payload + qos 0, which the consumer treats as a cleared (tombstone) msg.
91140
DefaultTbQueueMsgHeaders headers = new DefaultTbQueueMsgHeaders();

‎application/src/test/java/org/thingsboard/mqtt/broker/util/InitLoadUtilTest.java‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
/**
2+
* Copyright © 2016-2026 The Thingsboard Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package org.thingsboard.mqtt.broker.util;
17+
18+
import org.junit.Test;
19+
import org.thingsboard.mqtt.broker.exception.QueuePersistenceException;
20+
21+
import java.util.concurrent.atomic.AtomicInteger;
22+
23+
import static org.junit.Assert.assertEquals;
24+
import static org.junit.Assert.assertSame;
25+
import static org.junit.Assert.assertThrows;
26+
27+
public class InitLoadUtilTest {
28+
29+
@Test
30+
public void givenTransientFailures_whenPersistMarkerWithRetry_thenRetriesUntilSuccess() throws Exception {
31+
AtomicInteger attempts = new AtomicInteger();
32+
33+
InitLoadUtil.persistMarkerWithRetry("test", () -> {
34+
if (attempts.incrementAndGet() < 3) {
35+
throw new QueuePersistenceException("This server does not host this topic-partition.");
36+
}
37+
}, 5, 0);
38+
39+
assertEquals(3, attempts.get());
40+
}
41+
42+
@Test
43+
public void givenPersistentFailure_whenPersistMarkerWithRetry_thenThrowsLastErrorAfterMaxAttempts() {
44+
AtomicInteger attempts = new AtomicInteger();
45+
QueuePersistenceException error = new QueuePersistenceException("boom");
46+
47+
QueuePersistenceException thrown = assertThrows(QueuePersistenceException.class,
48+
() -> InitLoadUtil.persistMarkerWithRetry("test", () -> {
49+
attempts.incrementAndGet();
50+
throw error;
51+
}, 4, 0));
52+
53+
assertSame(error, thrown);
54+
assertEquals(4, attempts.get());
55+
}
56+
57+
@Test
58+
public void givenImmediateSuccess_whenPersistMarkerWithRetry_thenRunsOnce() throws Exception {
59+
AtomicInteger attempts = new AtomicInteger();
60+
61+
InitLoadUtil.persistMarkerWithRetry("test", attempts::incrementAndGet, 4, 0);
62+
63+
assertEquals(1, attempts.get());
64+
}
65+
}

0 commit comments

Comments
 (0)