com.fasterxml.jackson.core
diff --git a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java
index ab9c1e5f..66f8dd1d 100644
--- a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java
+++ b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java
@@ -1,9 +1,5 @@
package com.lingniu.ingest.inbound.mqtt.client;
-import com.hivemq.client.mqtt.datatypes.MqttQos;
-import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient;
-import com.hivemq.client.mqtt.mqtt3.Mqtt3Client;
-import com.hivemq.client.mqtt.mqtt3.message.connect.connack.Mqtt3ConnAck;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.pipeline.RawFrame;
import com.lingniu.ingest.core.dispatcher.Dispatcher;
@@ -15,27 +11,30 @@ import com.lingniu.ingest.identity.VehicleIdentitySource;
import com.lingniu.ingest.inbound.mqtt.config.MqttInboundProperties;
import com.lingniu.ingest.inbound.mqtt.model.MqttPayload;
import com.lingniu.ingest.inbound.mqtt.profile.MqttProfileRegistry;
+import org.fusesource.mqtt.client.Future;
+import org.fusesource.mqtt.client.FutureConnection;
+import org.fusesource.mqtt.client.MQTT;
+import org.fusesource.mqtt.client.Message;
+import org.fusesource.mqtt.client.QoS;
+import org.fusesource.mqtt.client.Topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.net.URI;
+import java.io.EOFException;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
/**
* MQTT 多 endpoint 生命周期管理。
*
- * 为每个 endpoint 创建一个 HiveMQ 异步客户端 → 连接 → 订阅 → 注册回调 →
+ *
为每个 endpoint 创建一个 fusesource MQTT FutureConnection → 连接 → 订阅 → 接收循环 →
* 回调里解析 payload → {@link Dispatcher#dispatch(RawFrame)}。
*
- *
Netty EventLoop 回调不做业务处理，虚拟线程由 Dispatcher 侧的 Disruptor 承接。
+ *
每个 endpoint 在独立虚拟线程中阻塞接收，业务处理由 Dispatcher 侧的 Disruptor 承接。
*/
public final class MqttEndpointManager implements AutoCloseable {
@@ -45,7 +44,9 @@ public final class MqttEndpointManager implements AutoCloseable {
private final MqttProfileRegistry profileRegistry;
private final VehicleIdentityResolver identityResolver;
private final Dispatcher dispatcher;
- private final List clients = new ArrayList<>();
+ private final List connections = new ArrayList<>();
+ private final List workers = new ArrayList<>();
+ private volatile boolean running;
public MqttEndpointManager(MqttInboundProperties props,
MqttProfileRegistry profileRegistry,
@@ -64,90 +65,110 @@ public final class MqttEndpointManager implements AutoCloseable {
}
public void start() {
+ running = true;
log.info("mqtt endpoint manager starting, endpoints={}", props.getEndpoints().size());
for (MqttInboundProperties.Endpoint ep : props.getEndpoints()) {
+ Thread worker = Thread.ofVirtual()
+ .name("mqtt-endpoint-" + safeName(ep.getName()))
+ .start(() -> runEndpoint(ep));
+ workers.add(worker);
+ }
+ }
+
+ private void runEndpoint(MqttInboundProperties.Endpoint ep) {
+ while (running && !Thread.currentThread().isInterrupted()) {
+ FutureConnection connection = null;
try {
- // 多 endpoint 互不影响：单个连接初始化失败会派发 operationalError，但不会阻止其他 endpoint。
log.info("mqtt endpoint [{}] initializing uri={} topic={} qos={} tls={}",
ep.getName(), ep.getUri(), ep.getTopic(), ep.getQos(), hasCustomTls(ep.getTls()));
- Mqtt3AsyncClient client = buildClient(ep);
- log.info("mqtt endpoint [{}] client built", ep.getName());
- clients.add(client);
- connectAndSubscribe(client, ep);
- } catch (Exception e) {
- log.error("mqtt endpoint [{}] init failed, skipped", ep.getName(), e);
- publishOperationalError(ep, "init", e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage());
- }
- }
- }
-
- private Mqtt3AsyncClient buildClient(MqttInboundProperties.Endpoint ep) {
- URI uri = URI.create(ep.getUri());
- int port = uri.getPort() > 0 ? uri.getPort() : defaultPort(uri.getScheme());
- var builder = Mqtt3Client.builder()
- .identifier(ep.getClientId() == null || ep.getClientId().isBlank()
- ? "lingniu-" + UUID.randomUUID() : ep.getClientId())
- .serverHost(uri.getHost())
- .serverPort(port)
- .automaticReconnectWithDefaultConfig();
- if (ep.getConnectionTimeoutSeconds() > 0) {
- builder = builder.transportConfig()
- .serverHost(uri.getHost())
- .serverPort(port)
- .socketConnectTimeout(ep.getConnectionTimeoutSeconds(), TimeUnit.SECONDS)
- .mqttConnectTimeout(ep.getConnectionTimeoutSeconds(), TimeUnit.SECONDS)
- .applyTransportConfig();
- }
-
- if ("ssl".equalsIgnoreCase(uri.getScheme()) || "tls".equalsIgnoreCase(uri.getScheme())
- || "mqtts".equalsIgnoreCase(uri.getScheme())) {
- if (hasCustomTls(ep.getTls())) {
- builder = builder.sslConfig(MqttTlsSupport.build(ep.getTls()));
- } else {
- builder = builder.sslWithDefaultConfig();
- }
- }
- return builder.buildAsync();
- }
-
- private void connectAndSubscribe(Mqtt3AsyncClient client, MqttInboundProperties.Endpoint ep) {
- var connectBuilder = client.connectWith()
- .cleanSession(ep.isCleanSession());
- if (ep.getKeepAliveSeconds() > 0) {
- connectBuilder = connectBuilder.keepAlive(ep.getKeepAliveSeconds());
- }
- if (ep.getUsername() != null && !ep.getUsername().isBlank()) {
- connectBuilder = connectBuilder.simpleAuth()
- .username(ep.getUsername())
- .password(ep.getPassword() == null ? new byte[0] : ep.getPassword().getBytes())
- .applySimpleAuth();
- }
- log.info("mqtt endpoint [{}] connect sending", ep.getName());
- CompletableFuture timeoutProbe = connectBuilder.send()
- .orTimeout(Math.max(15, ep.getConnectionTimeoutSeconds() + 5L), TimeUnit.SECONDS);
- timeoutProbe.whenComplete((ack, err) -> {
- if (err != null) {
- log.error("mqtt endpoint [{}] connect failed", ep.getName(), err);
- publishOperationalError(ep, "connect", err.getMessage() == null ? err.getClass().getSimpleName() : err.getMessage());
+ MQTT mqtt = buildClient(ep);
+ connection = mqtt.futureConnection();
+ connections.add(connection);
+ log.info("mqtt endpoint [{}] connect sending", ep.getName());
+ connection.connect().await();
+ log.info("mqtt endpoint [{}] connected", ep.getName());
+ connection.subscribe(new Topic[]{new Topic(ep.getTopic(), mapQos(ep.getQos()))}).await();
+ log.info("mqtt endpoint [{}] subscribed topic={} qos={}", ep.getName(), ep.getTopic(), ep.getQos());
+ receiveLoop(ep, connection);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
return;
+ } catch (Exception e) {
+ if (isPeerDisconnect(e)) {
+ log.warn("mqtt endpoint [{}] peer disconnected, reconnecting: {}", ep.getName(), e.getMessage());
+ } else {
+ log.error("mqtt endpoint [{}] connect/receive failed", ep.getName(), e);
+ publishOperationalError(ep, "connect",
+ e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage());
+ }
+ sleepBeforeReconnect(ep);
+ } finally {
+ if (connection != null) {
+ connections.remove(connection);
+ try {
+ connection.disconnect().await();
+ } catch (Exception ignored) {
+ // best-effort close
+ }
+ }
}
- log.info("mqtt endpoint [{}] connected code={}", ep.getName(), ack.getReturnCode());
- client.subscribeWith()
- .topicFilter(ep.getTopic())
- .qos(mapQos(ep.getQos()))
- .callback(publish -> onMessage(ep, publish.getTopic().toString(), publish.getPayloadAsBytes()))
- .send()
- .whenComplete((subAck, subErr) -> {
- if (subErr != null) {
- log.error("mqtt endpoint [{}] subscribe failed topic={}", ep.getName(), ep.getTopic(), subErr);
- publishOperationalError(ep, "subscribe",
- subErr.getMessage() == null ? subErr.getClass().getSimpleName() : subErr.getMessage());
- } else {
- log.info("mqtt endpoint [{}] subscribed topic={} qos={}",
- ep.getName(), ep.getTopic(), ep.getQos());
- }
- });
- });
+ }
+ }
+
+ private MQTT buildClient(MqttInboundProperties.Endpoint ep) throws Exception {
+ MQTT mqtt = new MQTT();
+ mqtt.setHost(ep.getUri());
+ if (ep.getClientId() != null && !ep.getClientId().isBlank()) {
+ mqtt.setClientId(ep.getClientId());
+ }
+ mqtt.setCleanSession(ep.isCleanSession());
+ if (ep.getUsername() != null && !ep.getUsername().isBlank()) {
+ mqtt.setUserName(ep.getUsername());
+ mqtt.setPassword(ep.getPassword() == null ? "" : ep.getPassword());
+ }
+ mqtt.setConnectAttemptsMax(1);
+ mqtt.setReconnectAttemptsMax(0);
+ if (ep.getKeepAliveSeconds() > 0) {
+ mqtt.setKeepAlive((short) ep.getKeepAliveSeconds());
+ }
+ mqtt.setVersion("3.1.1");
+ if (hasCustomTls(ep.getTls())) {
+ mqtt.setSslContext(MqttTlsSupport.buildSslContext(ep.getTls()));
+ }
+ return mqtt;
+ }
+
+ private void receiveLoop(MqttInboundProperties.Endpoint ep, FutureConnection connection) throws Exception {
+ while (running && !Thread.currentThread().isInterrupted()) {
+ Future futureMessage = connection.receive();
+ Message message = futureMessage.await();
+ try {
+ message.ack();
+ onMessage(ep, message.getTopic(), message.getPayload());
+ } catch (Exception e) {
+ log.warn("mqtt endpoint [{}] message ack/dispatch failed topic={} len={}",
+ ep.getName(), message.getTopic(), message.getPayload() == null ? 0 : message.getPayload().length, e);
+ }
+ }
+ }
+
+ private boolean isPeerDisconnect(Exception e) {
+ Throwable current = e;
+ while (current != null) {
+ if (current instanceof EOFException) {
+ return true;
+ }
+ current = current.getCause();
+ }
+ return false;
+ }
+
+ private void sleepBeforeReconnect(MqttInboundProperties.Endpoint ep) {
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
}
private void onMessage(MqttInboundProperties.Endpoint ep, String topic, byte[] payload) {
@@ -298,19 +319,11 @@ public final class MqttEndpointManager implements AutoCloseable {
return value.replace("\\", "\\\\").replace("\"", "\\\"");
}
- private static MqttQos mapQos(int level) {
+ private static QoS mapQos(int level) {
return switch (level) {
- case 0 -> MqttQos.AT_MOST_ONCE;
- case 2 -> MqttQos.EXACTLY_ONCE;
- default -> MqttQos.AT_LEAST_ONCE;
- };
- }
-
- private static int defaultPort(String scheme) {
- if (scheme == null) return 1883;
- return switch (scheme.toLowerCase()) {
- case "ssl", "tls", "mqtts" -> 8883;
- default -> 1883;
+ case 0 -> QoS.AT_MOST_ONCE;
+ case 2 -> QoS.EXACTLY_ONCE;
+ default -> QoS.AT_LEAST_ONCE;
};
}
@@ -318,6 +331,13 @@ public final class MqttEndpointManager implements AutoCloseable {
return value != null && value.trim().length() == 17;
}
+ private static String safeName(String value) {
+ if (value == null || value.isBlank()) {
+ return "unnamed";
+ }
+ return value.replaceAll("[^a-zA-Z0-9_.-]", "_");
+ }
+
private record IdentityResolution(VehicleIdentity identity, String errorMessage) {}
private static boolean hasCustomTls(MqttInboundProperties.Tls tls) {
@@ -331,13 +351,17 @@ public final class MqttEndpointManager implements AutoCloseable {
@Override
public void close() {
- for (Mqtt3AsyncClient c : clients) {
+ running = false;
+ for (FutureConnection connection : List.copyOf(connections)) {
try {
- c.disconnect();
+ connection.disconnect().await();
} catch (Exception ignored) {
// best-effort
}
}
- log.info("mqtt endpoint manager stopped, clients={}", clients.size());
+ for (Thread worker : workers) {
+ worker.interrupt();
+ }
+ log.info("mqtt endpoint manager stopped, connections={}", connections.size());
}
}
diff --git a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttTlsSupport.java b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttTlsSupport.java
index 0135da2b..527b3d1d 100644
--- a/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttTlsSupport.java
+++ b/modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttTlsSupport.java
@@ -1,18 +1,18 @@
package com.lingniu.ingest.inbound.mqtt.client;
-import com.hivemq.client.mqtt.MqttClientSslConfig;
-import com.hivemq.client.mqtt.MqttClientSslConfigBuilder;
import com.lingniu.ingest.inbound.mqtt.config.MqttInboundProperties;
import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManagerFactory;
-import java.io.ByteArrayInputStream;
+import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.KeyFactory;
import java.security.KeyStore;
import java.security.PrivateKey;
+import java.security.SecureRandom;
import java.security.cert.Certificate;
import java.security.cert.CertificateFactory;
import java.security.spec.PKCS8EncodedKeySpec;
@@ -25,74 +25,93 @@ public final class MqttTlsSupport {
private MqttTlsSupport() {}
- public static MqttClientSslConfig build(MqttInboundProperties.Tls tls) {
+ public static SSLContext buildSslContext(MqttInboundProperties.Tls tls) {
try {
- MqttClientSslConfigBuilder builder = MqttClientSslConfig.builder();
+ TrustManagerFactory tmf = null;
+ KeyManagerFactory kmf = null;
if (tls != null && hasText(tls.getCaPem())) {
- builder.trustManagerFactory(trustManagerFactory(Path.of(tls.getCaPem())));
+ tmf = trustManagerFactory(Path.of(tls.getCaPem()));
}
if (tls != null && hasText(tls.getClientPem()) && hasText(tls.getClientKey())) {
- builder.keyManagerFactory(keyManagerFactory(Path.of(tls.getClientPem()), Path.of(tls.getClientKey())));
+ kmf = keyManagerFactory(Path.of(tls.getClientPem()), Path.of(tls.getClientKey()));
}
- if (tls != null && !tls.isHostnameVerificationEnabled()) {
- builder.hostnameVerifier((hostname, session) -> true);
- }
- return builder.build();
+ SSLContext context = SSLContext.getInstance("TLSv1.2");
+ context.init(
+ kmf == null ? null : kmf.getKeyManagers(),
+ tmf == null ? null : tmf.getTrustManagers(),
+ new SecureRandom());
+ return context;
} catch (Exception e) {
throw new IllegalStateException("mqtt tls config build failed", e);
}
}
- private static TrustManagerFactory trustManagerFactory(Path caPem) throws Exception {
+ private static TrustManagerFactory trustManagerFactory(Path caPath) throws Exception {
KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
trustStore.load(null, null);
int i = 0;
- for (Certificate cert : readCertificates(caPem)) {
+ for (Certificate cert : readCertificates(caPath)) {
trustStore.setCertificateEntry("ca-" + i++, cert);
}
- TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ TrustManagerFactory tmf = TrustManagerFactory.getInstance("PKIX");
tmf.init(trustStore);
return tmf;
}
- private static KeyManagerFactory keyManagerFactory(Path clientPem, Path clientKey) throws Exception {
- List chain = readCertificates(clientPem);
- PrivateKey key = readPrivateKey(clientKey);
+ private static KeyManagerFactory keyManagerFactory(Path clientCertPath, Path clientKeyPath) throws Exception {
+ List chain = readCertificates(clientCertPath);
+ PrivateKey key = readPrivateKey(clientKeyPath);
KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());
keyStore.load(null, null);
char[] password = new char[0];
- keyStore.setKeyEntry("mqtt-client", key, password, chain.toArray(Certificate[]::new));
- KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ keyStore.setCertificateEntry("client", chain.getFirst());
+ keyStore.setKeyEntry("clientKey", key, password, chain.toArray(Certificate[]::new));
+ KeyManagerFactory kmf = KeyManagerFactory.getInstance("PKIX");
kmf.init(keyStore, password);
return kmf;
}
- private static List readCertificates(Path pem) throws Exception {
- String text = Files.readString(pem, StandardCharsets.UTF_8);
- List certificates = new ArrayList<>();
+ private static List readCertificates(Path path) throws Exception {
CertificateFactory factory = CertificateFactory.getInstance("X.509");
- for (String block : pemBlocks(text, "CERTIFICATE")) {
- try (ByteArrayInputStream in = new ByteArrayInputStream(Base64.getMimeDecoder().decode(block))) {
- certificates.add(factory.generateCertificate(in));
+ String text = readUtf8IfPossible(path);
+ if (text != null && text.contains("-----BEGIN CERTIFICATE-----")) {
+ List certificates = new ArrayList<>();
+ for (String block : pemBlocks(text, "CERTIFICATE")) {
+ certificates.add(factory.generateCertificate(
+ new java.io.ByteArrayInputStream(Base64.getMimeDecoder().decode(block))));
+ }
+ if (!certificates.isEmpty()) {
+ return certificates;
}
}
- if (certificates.isEmpty()) {
- throw new IllegalArgumentException("no certificate found in " + pem);
+ try (InputStream in = Files.newInputStream(path)) {
+ Collection certificates = factory.generateCertificates(in);
+ if (certificates.isEmpty()) {
+ throw new IllegalArgumentException("no certificate found in " + path);
+ }
+ return List.copyOf(certificates);
}
- return certificates;
}
- private static PrivateKey readPrivateKey(Path pem) throws Exception {
- String text = Files.readString(pem, StandardCharsets.UTF_8);
+ private static PrivateKey readPrivateKey(Path path) throws Exception {
+ String text = Files.readString(path, StandardCharsets.UTF_8);
List blocks = pemBlocks(text, "PRIVATE KEY");
if (blocks.isEmpty()) {
- throw new IllegalArgumentException("no PKCS#8 private key found in " + pem);
+ throw new IllegalArgumentException("no PKCS#8 private key found in " + path);
}
byte[] der = Base64.getMimeDecoder().decode(blocks.getFirst());
PKCS8EncodedKeySpec spec = new PKCS8EncodedKeySpec(der);
return KeyFactory.getInstance("RSA").generatePrivate(spec);
}
+ private static String readUtf8IfPossible(Path path) {
+ try {
+ return Files.readString(path, StandardCharsets.UTF_8);
+ } catch (Exception ignored) {
+ return null;
+ }
+ }
+
private static List pemBlocks(String text, String type) {
String begin = "-----BEGIN " + type + "-----";
String end = "-----END " + type + "-----";
diff --git a/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttEndpointManagerTest.java b/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttEndpointManagerTest.java
index 138c7392..a1c94db6 100644
--- a/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttEndpointManagerTest.java
+++ b/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttEndpointManagerTest.java
@@ -172,7 +172,7 @@ class MqttEndpointManagerTest {
.containsEntry("endpoint", "bad-endpoint")
.containsEntry("profile", "yutong")
.containsEntry("operationalError", "true")
- .containsEntry("phase", "init")
+ .containsEntry("phase", "connect")
.containsEntry("identityResolved", "false")
.containsEntry("identitySource", "UNKNOWN");
});
@@ -184,7 +184,7 @@ class MqttEndpointManagerTest {
.containsEntry("operationalError", "true")
.containsEntry("endpoint", "bad-endpoint")
.containsEntry("profile", "yutong")
- .containsEntry("phase", "init");
+ .containsEntry("phase", "connect");
});
batchExecutor.close();
diff --git a/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttTlsSupportTest.java b/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttTlsSupportTest.java
index ad41d63b..769b04c1 100644
--- a/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttTlsSupportTest.java
+++ b/modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttTlsSupportTest.java
@@ -1,6 +1,5 @@
package com.lingniu.ingest.inbound.mqtt;
-import com.hivemq.client.mqtt.MqttClientSslConfig;
import com.lingniu.ingest.inbound.mqtt.client.MqttTlsSupport;
import com.lingniu.ingest.inbound.mqtt.config.MqttInboundProperties;
import org.junit.jupiter.api.Test;
@@ -9,6 +8,8 @@ import org.junit.jupiter.api.io.TempDir;
import java.nio.file.Files;
import java.nio.file.Path;
+import javax.net.ssl.SSLContext;
+
import static org.assertj.core.api.Assertions.assertThat;
class MqttTlsSupportTest {
@@ -23,10 +24,9 @@ class MqttTlsSupportTest {
MqttInboundProperties.Tls tls = new MqttInboundProperties.Tls();
tls.setCaPem(ca.toString());
- MqttClientSslConfig config = MqttTlsSupport.build(tls);
+ SSLContext config = MqttTlsSupport.buildSslContext(tls);
- assertThat(config.getTrustManagerFactory()).isPresent();
- assertThat(config.getKeyManagerFactory()).isEmpty();
+ assertThat(config.getProtocol()).isEqualTo("TLSv1.2");
}
@Test
@@ -42,24 +42,22 @@ class MqttTlsSupportTest {
tls.setClientPem(cert.toString());
tls.setClientKey(key.toString());
- MqttClientSslConfig config = MqttTlsSupport.build(tls);
+ SSLContext config = MqttTlsSupport.buildSslContext(tls);
- assertThat(config.getTrustManagerFactory()).isPresent();
- assertThat(config.getKeyManagerFactory()).isPresent();
- assertThat(config.getHostnameVerifier()).isEmpty();
+ assertThat(config.getProtocol()).isEqualTo("TLSv1.2");
}
@Test
- void canDisableHostnameVerificationForLegacyMutualTlsBrokers() throws Exception {
+ void hostnameVerificationFlagDoesNotBlockSslContextCreation() throws Exception {
Path ca = tempDir.resolve("ca.pem");
Files.writeString(ca, CERT);
MqttInboundProperties.Tls tls = new MqttInboundProperties.Tls();
tls.setCaPem(ca.toString());
tls.setHostnameVerificationEnabled(false);
- MqttClientSslConfig config = MqttTlsSupport.build(tls);
+ SSLContext config = MqttTlsSupport.buildSslContext(tls);
- assertThat(config.getHostnameVerifier()).isPresent();
+ assertThat(config.getProtocol()).isEqualTo("TLSv1.2");
}
private static final String CERT = """
diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960FrameController.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960FrameController.java
index a6528476..c0cbdd5b 100644
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960FrameController.java
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Gb32960FrameController.java
@@ -27,6 +27,7 @@ import java.util.List;
*/
@RestController
@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true")
+@ConditionalOnProperty(prefix = "lingniu.ingest.event-history.api", name = "specialized-enabled", havingValue = "true")
@ConditionalOnBean(Gb32960DecodedFrameService.class)
@RequestMapping("/api/event-history/gb32960")
@Tag(name = "gb-32960-frame-controller", description = "GB32960 专用历史帧、业务快照、字段查询和字段字典接口。")
diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808LocationHistoryController.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808LocationHistoryController.java
index 2571d2e4..15bbb10e 100644
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808LocationHistoryController.java
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808LocationHistoryController.java
@@ -26,6 +26,7 @@ import java.util.List;
@RestController
@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true")
+@ConditionalOnProperty(prefix = "lingniu.ingest.event-history.api", name = "specialized-enabled", havingValue = "true")
@ConditionalOnBean(TdengineHistoryReader.class)
@RequestMapping("/api/event-history/jt808")
@Tag(name = "jt-808-location-controller", description = "JT808 位置历史分页查询接口。")
diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryController.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryController.java
index 2bf059b4..b6d79f94 100644
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryController.java
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/Jt808RawFrameHistoryController.java
@@ -25,6 +25,7 @@ import java.util.Locale;
@RestController
@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true")
+@ConditionalOnProperty(prefix = "lingniu.ingest.event-history.api", name = "specialized-enabled", havingValue = "true")
@ConditionalOnBean(TdengineHistoryReader.class)
@RequestMapping("/api/event-history/jt808")
@Tag(name = "jt-808-raw-frame-controller", description = "JT808 原始帧索引分页查询接口。")
diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/LocationHistoryController.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/LocationHistoryController.java
new file mode 100644
index 00000000..f39ac1a6
--- /dev/null
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/LocationHistoryController.java
@@ -0,0 +1,204 @@
+package com.lingniu.ingest.eventhistory;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
+import com.lingniu.ingest.tdenginehistory.TdengineLocationQuery;
+import com.lingniu.ingest.tdenginehistory.TdengineLocationRow;
+import com.lingniu.ingest.tdenginehistory.TdenginePage;
+import com.lingniu.ingest.tdenginehistory.TdenginePageCursor;
+import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.server.ResponseStatusException;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.List;
+import java.util.Locale;
+
+@RestController
+@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true")
+@ConditionalOnBean(TdengineHistoryReader.class)
+@RequestMapping("/api/event-history")
+@Tag(name = "location-history-controller", description = "通用位置历史分页查询接口。")
+public final class LocationHistoryController {
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ private final TdengineHistoryReader reader;
+
+ public LocationHistoryController(TdengineHistoryReader reader) {
+ if (reader == null) {
+ throw new IllegalArgumentException("reader must not be null");
+ }
+ this.reader = reader;
+ }
+
+ @GetMapping("/locations")
+ @Operation(
+ summary = "查询车辆位置历史",
+ description = "按协议、车辆标识和时间范围查询 TDengine 位置点。分页使用 cursorTs + cursorId，不使用 offset。")
+ public LocationPageResponse locations(
+ @Parameter(description = "协议名，例如 GB32960、JT808、MQTT_YUTONG、XINDA_PUSH。", required = true, example = "MQTT_YUTONG")
+ @RequestParam String protocol,
+ @Parameter(description = "内部车辆键。传入后优先使用。", example = "LMRKH9AC2R1004087")
+ @RequestParam(required = false) String vehicleKey,
+ @Parameter(description = "VIN。GB32960、宇通、信达通常可直接用 VIN。", example = "LMRKH9AC2R1004087")
+ @RequestParam(required = false) String vin,
+ @Parameter(description = "JT808 终端手机号或终端标识；JT808 会自动映射为 jt808:。", example = "g7gps")
+ @RequestParam(required = false) String phone,
+ @Parameter(description = "开始时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-29T13:00:00")
+ @RequestParam String dateFrom,
+ @Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-29T13:10:00")
+ @RequestParam String dateTo,
+ @Parameter(description = "排序方向。", example = "DESC")
+ @RequestParam(defaultValue = "DESC") TdengineQueryOrder order,
+ @Parameter(description = "返回位置点数量上限，最大 1000。", example = "100")
+ @RequestParam(defaultValue = "100") int limit,
+ @Parameter(description = "上一页返回的 nextCursor.ts。", example = "2026-06-29T05:00:01Z")
+ @RequestParam(required = false) String cursorTs,
+ @Parameter(description = "上一页返回的 nextCursor.id。", example = "location-xxx")
+ @RequestParam(required = false) String cursorId) throws IOException {
+ String normalizedProtocol = require(protocol, "protocol is required for location query").toUpperCase(Locale.ROOT);
+ QueryTimeRange range = QueryTimeRange.parse(dateFrom, dateTo);
+ TdenginePage page = reader.queryLocations(new TdengineLocationQuery(
+ normalizedProtocol,
+ resolveVehicleKey(normalizedProtocol, vehicleKey, vin, phone),
+ range.eventTimeFrom(),
+ range.eventTimeTo().plusMillis(1),
+ order,
+ limit,
+ cursor(cursorTs, cursorId)));
+ return LocationPageResponse.from(page);
+ }
+
+ private static String resolveVehicleKey(String protocol, String vehicleKey, String vin, String phone) {
+ String explicitVehicleKey = trimToNull(vehicleKey);
+ if (explicitVehicleKey != null) {
+ return explicitVehicleKey;
+ }
+ String vinValue = trimToNull(vin);
+ if (vinValue != null) {
+ return vinValue;
+ }
+ String phoneValue = trimToNull(phone);
+ if (phoneValue != null) {
+ if ("JT808".equals(protocol) && !phoneValue.startsWith("jt808:")) {
+ return "jt808:" + phoneValue;
+ }
+ return phoneValue;
+ }
+ throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "vehicleKey, vin or phone is required");
+ }
+
+ private static String require(String value, String message) {
+ String trimmed = trimToNull(value);
+ if (trimmed == null) {
+ throw new ResponseStatusException(HttpStatus.BAD_REQUEST, message);
+ }
+ return trimmed;
+ }
+
+ private static String trimToNull(String value) {
+ if (value == null || value.isBlank()) {
+ return null;
+ }
+ return value.trim();
+ }
+
+ private static TdenginePageCursor cursor(String cursorTs, String cursorId) {
+ boolean hasTs = cursorTs != null && !cursorTs.isBlank();
+ boolean hasId = cursorId != null && !cursorId.isBlank();
+ if (!hasTs && !hasId) {
+ return null;
+ }
+ if (!hasTs || !hasId) {
+ throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "cursorTs and cursorId must be provided together");
+ }
+ return new TdenginePageCursor(Instant.parse(cursorTs.trim()), cursorId.trim());
+ }
+
+ public record LocationPageResponse(
+ List items,
+ CursorResponse nextCursor) {
+
+ private static LocationPageResponse from(TdenginePage page) {
+ CursorResponse next = page.nextCursor()
+ .map(cursor -> new CursorResponse(cursor.ts().toString(), cursor.tieBreaker()))
+ .orElse(null);
+ return new LocationPageResponse(
+ page.items().stream().map(LocationResponse::from).toList(),
+ next);
+ }
+ }
+
+ public record CursorResponse(String ts, String id) {
+ }
+
+ public record LocationResponse(
+ String eventTime,
+ String receivedAt,
+ String factId,
+ String frameId,
+ String phone,
+ String vin,
+ String vehicleKey,
+ double longitude,
+ double latitude,
+ double altitudeM,
+ double speedKmh,
+ double directionDeg,
+ long alarmFlag,
+ long statusFlag,
+ Double totalMileageKm,
+ String rawUri,
+ String metadataJson,
+ String protocol) {
+
+ private static LocationResponse from(TdengineLocationRow row) {
+ return new LocationResponse(
+ row.ts().toString(),
+ row.receivedAt().toString(),
+ row.factId(),
+ row.frameId(),
+ row.phone(),
+ row.vin(),
+ row.vehicleKey(),
+ row.longitude(),
+ row.latitude(),
+ row.altitudeM(),
+ row.speedKmh(),
+ row.directionDeg(),
+ row.alarmFlag(),
+ row.statusFlag(),
+ row.totalMileageKm(),
+ rawUri(row),
+ row.metadataJson(),
+ row.protocol());
+ }
+
+ private static String rawUri(TdengineLocationRow row) {
+ if (row.rawUri() != null && !row.rawUri().isBlank()) {
+ return row.rawUri();
+ }
+ String metadataJson = row.metadataJson();
+ if (metadataJson == null || metadataJson.isBlank()) {
+ return row.rawUri();
+ }
+ try {
+ return OBJECT_MAPPER.readTree(metadataJson).path("rawArchiveUri").asText(row.rawUri());
+ } catch (JsonProcessingException ignored) {
+ return row.rawUri();
+ }
+ }
+ }
+}
diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java
index cf481fd5..efd8fb4c 100644
--- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java
+++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java
@@ -9,6 +9,7 @@ import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
import com.lingniu.ingest.eventhistory.Gb32960FrameController;
import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController;
import com.lingniu.ingest.eventhistory.Jt808RawFrameHistoryController;
+import com.lingniu.ingest.eventhistory.LocationHistoryController;
import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController;
import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
@@ -95,6 +96,7 @@ public class EventHistoryAutoConfiguration {
@Bean
@ConditionalOnBean(Gb32960DecodedFrameService.class)
+ @ConditionalOnProperty(prefix = "lingniu.ingest.event-history.api", name = "specialized-enabled", havingValue = "true")
@ConditionalOnMissingBean
public Gb32960FrameController gb32960FrameController(Gb32960DecodedFrameService service) {
return new Gb32960FrameController(service);
@@ -103,12 +105,21 @@ public class EventHistoryAutoConfiguration {
@Bean
@ConditionalOnBean(TdengineHistoryReader.class)
@ConditionalOnMissingBean
+ public LocationHistoryController locationHistoryController(TdengineHistoryReader reader) {
+ return new LocationHistoryController(reader);
+ }
+
+ @Bean
+ @ConditionalOnBean(TdengineHistoryReader.class)
+ @ConditionalOnProperty(prefix = "lingniu.ingest.event-history.api", name = "specialized-enabled", havingValue = "true")
+ @ConditionalOnMissingBean
public Jt808LocationHistoryController jt808LocationHistoryController(TdengineHistoryReader reader) {
return new Jt808LocationHistoryController(reader);
}
@Bean
@ConditionalOnBean(TdengineHistoryReader.class)
+ @ConditionalOnProperty(prefix = "lingniu.ingest.event-history.api", name = "specialized-enabled", havingValue = "true")
@ConditionalOnMissingBean
public Jt808RawFrameHistoryController jt808RawFrameHistoryController(TdengineHistoryReader reader) {
return new Jt808RawFrameHistoryController(reader);
diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/LocationHistoryControllerTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/LocationHistoryControllerTest.java
new file mode 100644
index 00000000..0004ea2c
--- /dev/null
+++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/LocationHistoryControllerTest.java
@@ -0,0 +1,113 @@
+package com.lingniu.ingest.eventhistory;
+
+import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader;
+import com.lingniu.ingest.tdenginehistory.TdengineLocationQuery;
+import com.lingniu.ingest.tdenginehistory.TdengineLocationRow;
+import com.lingniu.ingest.tdenginehistory.TdenginePage;
+import com.lingniu.ingest.tdenginehistory.TdenginePageCursor;
+import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder;
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class LocationHistoryControllerTest {
+
+ @Test
+ void queriesGenericLocationHistoryWithCursor() throws Exception {
+ AtomicReference captured = new AtomicReference<>();
+ TdengineHistoryReader reader = new StubReader(captured, new TdenginePage<>(
+ List.of(row("fact-1")),
+ Optional.of(new TdenginePageCursor(Instant.parse("2026-06-29T05:00:01Z"), "fact-1"))));
+ LocationHistoryController controller = new LocationHistoryController(reader);
+
+ LocationHistoryController.LocationPageResponse response = controller.locations(
+ "mqtt_yutong",
+ null,
+ "LMRKH9AC2R1004087",
+ null,
+ "2026-06-29T13:00:00",
+ "2026-06-29T13:10:00",
+ TdengineQueryOrder.DESC,
+ 100,
+ null,
+ null);
+
+ assertThat(captured.get().protocol()).isEqualTo("MQTT_YUTONG");
+ assertThat(captured.get().vehicleKey()).isEqualTo("LMRKH9AC2R1004087");
+ assertThat(response.items()).extracting(LocationHistoryController.LocationResponse::factId)
+ .containsExactly("fact-1");
+ assertThat(response.nextCursor()).isEqualTo(new LocationHistoryController.CursorResponse(
+ "2026-06-29T05:00:01Z", "fact-1"));
+ }
+
+ @Test
+ void mapsJt808PhoneToVehicleKey() throws Exception {
+ AtomicReference captured = new AtomicReference<>();
+ LocationHistoryController controller = new LocationHistoryController(
+ new StubReader(captured, new TdenginePage<>(List.of(), null)));
+
+ controller.locations(
+ "JT808",
+ null,
+ null,
+ "g7gps",
+ "2026-06-29T13:00:00",
+ "2026-06-29T13:10:00",
+ TdengineQueryOrder.DESC,
+ 100,
+ null,
+ null);
+
+ assertThat(captured.get().vehicleKey()).isEqualTo("jt808:g7gps");
+ }
+
+ private static TdengineLocationRow row(String factId) {
+ Instant ts = Instant.parse("2026-06-29T05:00:00Z");
+ return new TdengineLocationRow(
+ ts,
+ factId,
+ "frame-1",
+ ts,
+ 121.1,
+ 30.5,
+ 0.0,
+ 32.0,
+ 90.0,
+ 0,
+ 1,
+ null,
+ "",
+ "{\"rawArchiveUri\":\"archive://raw.bin\"}",
+ "MQTT_YUTONG",
+ "LMRKH9AC2R1004087",
+ "LMRKH9AC2R1004087",
+ "");
+ }
+
+ private record StubReader(AtomicReference captured,
+ TdenginePage page) implements TdengineHistoryReader {
+
+ @Override
+ public com.lingniu.ingest.tdenginehistory.TdenginePage queryRawFrames(
+ com.lingniu.ingest.tdenginehistory.TdengineRawFrameQuery query) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TdenginePage queryLocations(TdengineLocationQuery query) {
+ captured.set(query);
+ return page;
+ }
+
+ @Override
+ public com.lingniu.ingest.tdenginehistory.TdenginePage queryTelemetryFields(
+ com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldQuery query) {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java
index c643f54b..24bc4030 100644
--- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java
+++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java
@@ -9,6 +9,7 @@ import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService;
import com.lingniu.ingest.eventhistory.Gb32960FrameController;
import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController;
import com.lingniu.ingest.eventhistory.Jt808RawFrameHistoryController;
+import com.lingniu.ingest.eventhistory.LocationHistoryController;
import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController;
import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper;
import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
@@ -58,6 +59,7 @@ class EventHistoryAutoConfigurationTest {
contextRunner
.withPropertyValues(
"lingniu.ingest.event-history.enabled=true",
+ "lingniu.ingest.event-history.api.specialized-enabled=true",
"lingniu.ingest.sink.archive.path=/tmp/lingniu-test-archive")
.withBean(Gb32960MessageDecoder.class, () -> mock(Gb32960MessageDecoder.class))
.withBean(SinkArchiveProperties.class, SinkArchiveProperties::new)
@@ -73,6 +75,7 @@ class EventHistoryAutoConfigurationTest {
.withConfiguration(AutoConfigurations.of(EventHistoryAutoConfiguration.class))
.withPropertyValues(
"lingniu.ingest.event-history.enabled=true",
+ "lingniu.ingest.event-history.api.specialized-enabled=true",
"lingniu.ingest.sink.archive.path=/tmp/lingniu-test-archive")
.withBean(Gb32960MessageDecoder.class, () -> mock(Gb32960MessageDecoder.class))
.withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
@@ -87,16 +90,33 @@ class EventHistoryAutoConfigurationTest {
@Test
void createsJt808LocationHistoryControllerWhenTdengineReaderExists() {
contextRunner
- .withPropertyValues("lingniu.ingest.event-history.enabled=true")
+ .withPropertyValues(
+ "lingniu.ingest.event-history.enabled=true",
+ "lingniu.ingest.event-history.api.specialized-enabled=true")
.withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
.run(context -> assertThat(context).hasSingleBean(Jt808LocationHistoryController.class));
}
@Test
- void createsJt808RawFrameHistoryControllerWhenTdengineReaderExists() {
+ void createsGenericLocationAndTelemetryControllersWhenTdengineReaderExists() {
contextRunner
.withPropertyValues("lingniu.ingest.event-history.enabled=true")
.withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
+ .run(context -> {
+ assertThat(context).hasSingleBean(LocationHistoryController.class);
+ assertThat(context).hasSingleBean(TelemetryFieldHistoryController.class);
+ assertThat(context).doesNotHaveBean(Jt808LocationHistoryController.class);
+ assertThat(context).doesNotHaveBean(Jt808RawFrameHistoryController.class);
+ });
+ }
+
+ @Test
+ void createsJt808RawFrameHistoryControllerWhenTdengineReaderExists() {
+ contextRunner
+ .withPropertyValues(
+ "lingniu.ingest.event-history.enabled=true",
+ "lingniu.ingest.event-history.api.specialized-enabled=true")
+ .withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class))
.run(context -> assertThat(context).hasSingleBean(Jt808RawFrameHistoryController.class));
}
diff --git a/pom.xml b/pom.xml
index d4af1051..2b6417af 100644
--- a/pom.xml
+++ b/pom.xml
@@ -65,7 +65,7 @@
2.2.0
1.13.6
1.42.1
- 1.3.3
+ 1.12
1.1.3
3.8.4
2.8.17
@@ -307,9 +307,9 @@
${kafka.version}