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 #14425 from volodymyr-babak/edge-zombie-session-fix 

Prevent dead Kafka consumer groups causing Edge sync failures
pull/14443/head
Viacheslav Klimov 10 months ago
committed by GitHub
parent
e3d1600549 3a0d083610
commit
93b625474f
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
2 changed files with 92 additions and 26 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 91
      application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
  2. 27
      application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java

91
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
View File

@ -69,6 +69,7 @@ import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap; import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
@ -82,6 +83,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer; import java.util.function.Consumer;
import java.util.function.Function;
import static org.thingsboard.server.service.state.DefaultDeviceStateService.ACTIVITY_STATE; import static org.thingsboard.server.service.state.DefaultDeviceStateService.ACTIVITY_STATE;
import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAST_CONNECT_TIME; import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAST_CONNECT_TIME;
@ -94,6 +96,7 @@ import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAS
public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase implements EdgeRpcService { public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase implements EdgeRpcService {
private final ConcurrentMap<EdgeId, EdgeGrpcSession> sessions = new ConcurrentHashMap<>(); private final ConcurrentMap<EdgeId, EdgeGrpcSession> sessions = new ConcurrentHashMap<>();
private final ConcurrentMap<UUID, EdgeGrpcSession> sessionsById = new ConcurrentHashMap<>();
private final ConcurrentMap<EdgeId, Lock> sessionNewEventsLocks = new ConcurrentHashMap<>(); private final ConcurrentMap<EdgeId, Lock> sessionNewEventsLocks = new ConcurrentHashMap<>();
private final Map<EdgeId, Boolean> sessionNewEvents = new HashMap<>(); private final Map<EdgeId, Boolean> sessionNewEvents = new HashMap<>();
private final ConcurrentMap<EdgeId, ScheduledFuture<?>> sessionEdgeEventChecks = new ConcurrentHashMap<>(); private final ConcurrentMap<EdgeId, ScheduledFuture<?>> sessionEdgeEventChecks = new ConcurrentHashMap<>();
@ -283,6 +286,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
destroySession(session); destroySession(session);
session.cleanUp(); session.cleanUp();
sessions.remove(edgeId); sessions.remove(edgeId);
sessionsById.remove(session.getSessionId());
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock()); final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock());
newEventLock.lock(); newEventLock.lock();
try { try {
@ -332,9 +336,15 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
TenantId tenantId = edge.getTenantId(); TenantId tenantId = edge.getTenantId();
log.info("[{}][{}] edge [{}] connected successfully.", tenantId, edgeGrpcSession.getSessionId(), edgeId); log.info("[{}][{}] edge [{}] connected successfully.", tenantId, edgeGrpcSession.getSessionId(), edgeId);
if (sessions.containsKey(edgeId)) { if (sessions.containsKey(edgeId)) {
destroySession(sessions.get(edgeId)); EdgeGrpcSession existing = sessions.get(edgeId);
if (existing != null) {
log.info("[{}][{}] Replacing existing session [{}] for edge [{}]", tenantId, edgeGrpcSession.getSessionId(), existing.getSessionId(), edgeId);
destroySession(existing);
sessionsById.remove(existing.getSessionId());
}
} }
sessions.put(edgeId, edgeGrpcSession); sessions.put(edgeId, edgeGrpcSession);
sessionsById.put(edgeGrpcSession.getSessionId(), edgeGrpcSession);
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock()); final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock());
newEventLock.lock(); newEventLock.lock();
try { try {
@ -492,9 +502,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
private void onEdgeDisconnect(Edge edge, UUID sessionId) { private void onEdgeDisconnect(Edge edge, UUID sessionId) {
EdgeId edgeId = edge.getId(); EdgeId edgeId = edge.getId();
log.info("[{}][{}] edge disconnected!", edgeId, sessionId); log.info("[{}][{}] edge disconnected!", edgeId, sessionId);
EdgeGrpcSession toRemove = sessions.get(edgeId); EdgeGrpcSession current = sessions.get(edgeId);
if (toRemove.getSessionId().equals(sessionId)) { if (current != null && current.getSessionId().equals(sessionId)) {
toRemove = sessions.remove(edgeId); EdgeGrpcSession toRemove = sessions.remove(edgeId);
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock()); final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock());
newEventLock.lock(); newEventLock.lock();
try { try {
@ -503,6 +513,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
newEventLock.unlock(); newEventLock.unlock();
} }
destroySession(toRemove); destroySession(toRemove);
sessionsById.remove(sessionId);
TenantId tenantId = toRemove.getEdge().getTenantId(); TenantId tenantId = toRemove.getEdge().getTenantId();
save(tenantId, edgeId, ACTIVITY_STATE, false); save(tenantId, edgeId, ACTIVITY_STATE, false);
long lastDisconnectTs = System.currentTimeMillis(); long lastDisconnectTs = System.currentTimeMillis();
@ -510,7 +521,18 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
pushRuleEngineMessage(toRemove.getEdge().getTenantId(), edge, lastDisconnectTs, TbMsgType.DISCONNECT_EVENT); pushRuleEngineMessage(toRemove.getEdge().getTenantId(), edge, lastDisconnectTs, TbMsgType.DISCONNECT_EVENT);
cancelScheduleEdgeEventsCheck(edgeId); cancelScheduleEdgeEventsCheck(edgeId);
} else { } else {
log.debug("[{}] edge session [{}] is not available anymore, nothing to remove. most probably this session is already outdated!", edgeId, sessionId); log.info("[{}] edge session [{}] is not current anymore. Attempting to destroy it by sessionId.", edgeId, sessionId);
EdgeGrpcSession stale = sessionsById.remove(sessionId);
if (stale != null) {
try {
destroySession(stale);
log.info("[{}][{}] Successfully destroyed stale session for edge [{}]", stale.getTenantId(), sessionId, edgeId);
} catch (Exception e) {
log.warn("[{}][{}] Failed to destroy stale session for edge [{}]", stale.getTenantId(), sessionId, edgeId, e);
}
} else {
log.debug("[{}] No session found by sessionId [{}] to destroy", edgeId, sessionId);
}
} }
edgeIdServiceIdCache.evict(edgeId); edgeIdServiceIdCache.evict(edgeId);
} }
@ -522,6 +544,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
session.getTenantId(), session.getEdge().getId(), session.getEdge().getName(), session.getSessionId()); session.getTenantId(), session.getEdge().getId(), session.getEdge().getName(), session.getSessionId());
zombieSessions.add(session); zombieSessions.add(session);
} }
} catch (Exception e) {
log.warn("[{}][{}] Exception during session destroy for edge [{}] with session id [{}]",
session.getTenantId(), session.getEdge().getId(), session.getEdge().getName(), session.getSessionId(), e);
} }
} }
@ -631,25 +656,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
private void cleanupZombieSessions() { private void cleanupZombieSessions() {
try { try {
List<EdgeId> toRemove = new ArrayList<>(); tryToDestroyZombieSessions(getZombieSessions(sessions.values()), s -> sessions.remove(s.getEdge().getId()));
for (EdgeGrpcSession session : sessions.values()) { tryToDestroyZombieSessions(getZombieSessions(sessionsById.values()), s -> sessionsById.remove(s.getSessionId()));
if (session instanceof KafkaEdgeGrpcSession kafkaSession &&
!kafkaSession.isConnected() &&
kafkaSession.getConsumer() != null &&
kafkaSession.getConsumer().getConsumer() != null &&
!kafkaSession.getConsumer().getConsumer().isStopped()) {
toRemove.add(kafkaSession.getEdge().getId());
}
}
for (EdgeId edgeId : toRemove) {
log.info("[{}] Destroying session for edge because edge is not connected", edgeId);
EdgeGrpcSession removed = sessions.get(edgeId);
if (removed instanceof KafkaEdgeGrpcSession kafkaSession) {
if (kafkaSession.destroy()) {
sessions.remove(edgeId);
}
}
}
zombieSessions.removeIf(zombie -> { zombieSessions.removeIf(zombie -> {
if (zombie.destroy()) { if (zombie.destroy()) {
log.info("[{}][{}] Successfully cleaned up zombie session [{}] for edge [{}].", log.info("[{}][{}] Successfully cleaned up zombie session [{}] for edge [{}].",
@ -666,4 +675,38 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
} }
} }
private List<EdgeGrpcSession> getZombieSessions(Collection<EdgeGrpcSession> sessions) {
List<EdgeGrpcSession> result = new ArrayList<>();
for (EdgeGrpcSession session : sessions) {
if (isKafkaSessionAndZombie(session)) {
result.add(session);
}
}
return result;
}
private void tryToDestroyZombieSessions(List<EdgeGrpcSession> sessionsToRemove, Function<EdgeGrpcSession, EdgeGrpcSession> removeFunc) {
for (EdgeGrpcSession toRemove : sessionsToRemove) {
log.info("[{}] Destroying session for edge because edge is not connected", toRemove.getEdge().getId());
if (toRemove.destroy()) {
removeFunc.apply(toRemove);
}
}
}
private boolean isKafkaSessionAndZombie(EdgeGrpcSession session) {
if (session instanceof KafkaEdgeGrpcSession kafkaSession) {
log.debug("[{}] kafkaSession.isConnected() = {}, kafkaSession.getConsumer().getConsumer().isStopped() = {}",
kafkaSession.getEdge().getId(),
kafkaSession.isConnected(),
kafkaSession.getConsumer() != null ? kafkaSession.getConsumer().getConsumer() != null ? kafkaSession.getConsumer().getConsumer().isStopped() : null : null);
return !kafkaSession.isConnected() &&
kafkaSession.getConsumer() != null &&
kafkaSession.getConsumer().getConsumer() != null &&
!kafkaSession.getConsumer().getConsumer().isStopped();
}
return false;
}
} }

27
application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java
View File

@ -101,8 +101,21 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession {
@Override @Override
public ListenableFuture<Boolean> processEdgeEvents() { public ListenableFuture<Boolean> processEdgeEvents() {
if (!isConnected() || isSyncInProgress() || isHighPriorityProcessing) {
log.warn("[{}][{}] Session is not ready (connected={}, syncInProgress={}, highPriority={}), skip starting edge event consumer",
tenantId, edge != null ? edge.getId() : null, isConnected(), isSyncInProgress(), isHighPriorityProcessing);
return Futures.immediateFuture(Boolean.FALSE);
}
if (consumer == null || (consumer.getConsumer() != null && consumer.getConsumer().isStopped())) { if (consumer == null || (consumer.getConsumer() != null && consumer.getConsumer().isStopped())) {
try { try {
if (consumerExecutor != null && !consumerExecutor.isShutdown()) {
try {
consumerExecutor.shutdown();
awaitConsumerTermination();
} catch (Exception e) {
log.warn("[{}][{}] Failed to shutdown previous consumer executor", tenantId, edge.getId(), e);
}
}
this.consumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("edge-event-consumer")); this.consumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("edge-event-consumer"));
this.consumer = QueueConsumerManager.<TbProtoQueueMsg<ToEdgeEventNotificationMsg>>builder() this.consumer = QueueConsumerManager.<TbProtoQueueMsg<ToEdgeEventNotificationMsg>>builder()
.name("TB Edge events [" + edge.getId() + "]") .name("TB Edge events [" + edge.getId() + "]")
@ -133,6 +146,7 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession {
public boolean destroy() { public boolean destroy() {
try { try {
if (consumer != null) { if (consumer != null) {
log.info("[{}][{}] Stopping edge event consumer...", tenantId, edge != null ? edge.getId() : null);
consumer.stop(); consumer.stop();
} }
} catch (Exception e) { } catch (Exception e) {
@ -141,16 +155,25 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession {
} }
consumer = null; consumer = null;
try { try {
if (consumerExecutor != null) { if (consumerExecutor != null && !consumerExecutor.isShutdown()) {
consumerExecutor.shutdown(); consumerExecutor.shutdown();
awaitConsumerTermination();
} }
} catch (Exception e) { } catch (Exception e) {
log.warn("[{}][{}] Failed to shutdown consumer executor", tenantId, edge.getId(), e); log.warn("[{}][{}] Failed to shutdown edge event consumer executor", tenantId, edge.getId(), e);
return false; return false;
} }
return true; return true;
} }
private void awaitConsumerTermination() {
try {
consumerExecutor.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS);
} catch (InterruptedException ie) {
log.warn("[{}][{}] Interrupted while awaiting consumer executor termination", tenantId, edge.getId());
}
}
@Override @Override
public void cleanUp() { public void cleanUp() {
String topic = topicService.buildEdgeEventNotificationsTopicPartitionInfo(tenantId, edge.getId()).getTopic(); String topic = topicService.buildEdgeEventNotificationsTopicPartitionInfo(tenantId, edge.getId()).getTopic();

Write Preview
Loading…
Cancel
Save