feat: make gb32960 archive history query production ready
This commit is contained in:
kkfluous
60
modules/sinks/event-file-store/pom.xml
Normal file
60
modules/sinks/event-file-store/pom.xml
Normal file
@@ -0,0 +1,60 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>com.lingniu.ingest</groupId>
|
||||
<artifactId>lingniu-vehicle-ingest</artifactId>
|
||||
<version>0.1.0-SNAPSHOT</version>
|
||||
<relativePath>../../../pom.xml</relativePath>
|
||||
</parent>
|
||||
<artifactId>event-file-store</artifactId>
|
||||
<name>event-file-store</name>
|
||||
<description>车辆事件文件型明细库：Parquet 存储 + DuckDB 时间顺序查询。</description>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>com.lingniu.ingest</groupId>
|
||||
<artifactId>ingest-api</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-autoconfigure</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-configuration-processor</artifactId>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.duckdb</groupId>
|
||||
<artifactId>duckdb_jdbc</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.core</groupId>
|
||||
<artifactId>jackson-databind</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.datatype</groupId>
|
||||
<artifactId>jackson-datatype-jsr310</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.junit.jupiter</groupId>
|
||||
<artifactId>junit-jupiter</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.assertj</groupId>
|
||||
<artifactId>assertj-core</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
@@ -0,0 +1,485 @@
|
||||
package com.lingniu.ingest.eventfilestore;
|
||||
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.lingniu.ingest.api.ProtocolId;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.StandardCopyOption;
|
||||
import java.sql.Connection;
|
||||
import java.sql.DriverManager;
|
||||
import java.sql.PreparedStatement;
|
||||
import java.sql.ResultSet;
|
||||
import java.sql.SQLException;
|
||||
import java.sql.Statement;
|
||||
import java.time.Instant;
|
||||
import java.time.LocalDate;
|
||||
import java.time.ZoneId;
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
public final class DuckDbParquetEventFileStore implements EventFileStore {
|
||||
|
||||
private static final TypeReference<Map<String, String>> STRING_MAP =
|
||||
new TypeReference<>() {};
|
||||
private static final String HEADER_VERSION = "event-records-v1";
|
||||
|
||||
private final Path root;
|
||||
private final Path indexPath;
|
||||
private final ZoneId partitionZone;
|
||||
private final ObjectMapper objectMapper;
|
||||
private volatile boolean indexInitialized;
|
||||
|
||||
public DuckDbParquetEventFileStore(Path root, ZoneId partitionZone) {
|
||||
this(root, partitionZone, new ObjectMapper());
|
||||
}
|
||||
|
||||
public DuckDbParquetEventFileStore(Path root, ZoneId partitionZone, ObjectMapper objectMapper) {
|
||||
if (root == null) {
|
||||
throw new IllegalArgumentException("root must not be null");
|
||||
}
|
||||
this.root = root.toAbsolutePath();
|
||||
this.indexPath = this.root.resolve("events.duckdb");
|
||||
this.partitionZone = partitionZone == null ? ZoneId.of("Asia/Shanghai") : partitionZone;
|
||||
this.objectMapper = objectMapper == null ? new ObjectMapper() : objectMapper;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void appendAll(List<EventFileRecord> records) throws IOException {
|
||||
if (records == null || records.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
Map<Partition, List<EventFileRecord>> byPartition = new LinkedHashMap<>();
|
||||
for (EventFileRecord record : records) {
|
||||
Partition partition = partition(record);
|
||||
byPartition.computeIfAbsent(partition, ignored -> new ArrayList<>()).add(record);
|
||||
}
|
||||
for (Map.Entry<Partition, List<EventFileRecord>> entry : byPartition.entrySet()) {
|
||||
writePartition(entry.getKey(), entry.getValue());
|
||||
appendIndex(entry.getKey(), entry.getValue());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<EventFileRecord> query(EventFileQuery query) throws IOException {
|
||||
if (query.vin() != null) {
|
||||
return queryParquet(query);
|
||||
}
|
||||
ensureIndexInitialized();
|
||||
return queryIndex(query);
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized EventFileRecord findByRawArchiveUri(String rawArchiveUri) throws IOException {
|
||||
if (rawArchiveUri == null || rawArchiveUri.isBlank()) {
|
||||
return null;
|
||||
}
|
||||
ensureIndexInitialized();
|
||||
String sql = """
|
||||
SELECT event_id, protocol, event_type, vin,
|
||||
event_time_ms, ingest_time_ms,
|
||||
raw_archive_uri, metadata_json, payload_json
|
||||
FROM event_records
|
||||
WHERE raw_archive_uri = ?
|
||||
ORDER BY event_type = 'RAW_ARCHIVE' DESC, ingest_time_ms DESC, event_id DESC
|
||||
LIMIT 1
|
||||
""";
|
||||
try (Connection connection = DriverManager.getConnection(indexJdbcUrl());
|
||||
PreparedStatement ps = connection.prepareStatement(sql)) {
|
||||
ps.setString(1, rawArchiveUri);
|
||||
try (ResultSet rs = ps.executeQuery()) {
|
||||
return rs.next() ? record(rs) : null;
|
||||
}
|
||||
} catch (SQLException e) {
|
||||
throw new IOException("query duckdb event index by raw archive uri failed", e);
|
||||
}
|
||||
}
|
||||
|
||||
private List<EventFileRecord> queryParquet(EventFileQuery query) throws IOException {
|
||||
List<Path> parquetFiles = parquetFiles(query);
|
||||
if (parquetFiles.isEmpty()) {
|
||||
return List.of();
|
||||
}
|
||||
String fileList = toDuckDbList(parquetFiles);
|
||||
String order = query.order() == EventFileQuery.Order.DESC ? "DESC" : "ASC";
|
||||
List<String> predicates = new ArrayList<>();
|
||||
if (query.vin() != null) {
|
||||
predicates.add("vin = '" + escapeSql(query.vin()) + "'");
|
||||
}
|
||||
if (query.eventType() != null) {
|
||||
predicates.add("event_type = '" + escapeSql(query.eventType()) + "'");
|
||||
}
|
||||
if (query.eventTimeFrom() != null) {
|
||||
predicates.add("event_time_ms >= " + query.eventTimeFrom().toEpochMilli());
|
||||
}
|
||||
if (query.eventTimeTo() != null) {
|
||||
predicates.add("event_time_ms <= " + query.eventTimeTo().toEpochMilli());
|
||||
}
|
||||
String where = predicates.isEmpty() ? "" : "WHERE " + String.join(" AND ", predicates) + "\n";
|
||||
String sql = """
|
||||
SELECT event_id, protocol, event_type, vin,
|
||||
event_time_ms, event_time, ingest_time_ms, ingest_time,
|
||||
raw_archive_uri, metadata_json, payload_json
|
||||
FROM read_parquet(%s)
|
||||
%s
|
||||
ORDER BY event_time_ms %s, ingest_time_ms %s, event_id %s
|
||||
LIMIT %d
|
||||
""".formatted(fileList, where, order, order, order, query.limit());
|
||||
try (Connection connection = DriverManager.getConnection("jdbc:duckdb:");
|
||||
Statement statement = connection.createStatement();
|
||||
ResultSet rs = statement.executeQuery(sql)) {
|
||||
List<EventFileRecord> result = new ArrayList<>();
|
||||
while (rs.next()) {
|
||||
result.add(new EventFileRecord(
|
||||
rs.getString("event_id"),
|
||||
protocol(rs.getString("protocol")),
|
||||
rs.getString("event_type"),
|
||||
rs.getString("vin"),
|
||||
Instant.ofEpochMilli(rs.getLong("event_time_ms")),
|
||||
Instant.ofEpochMilli(rs.getLong("ingest_time_ms")),
|
||||
rs.getString("raw_archive_uri"),
|
||||
readMetadata(rs.getString("metadata_json")),
|
||||
rs.getString("payload_json")
|
||||
));
|
||||
}
|
||||
return result;
|
||||
} catch (SQLException e) {
|
||||
throw new IOException("query parquet event store failed", e);
|
||||
}
|
||||
}
|
||||
|
||||
private List<EventFileRecord> queryIndex(EventFileQuery query) throws IOException {
|
||||
String order = query.order() == EventFileQuery.Order.DESC ? "DESC" : "ASC";
|
||||
StringBuilder where = new StringBuilder("""
|
||||
WHERE protocol = ?
|
||||
AND partition_date BETWEEN CAST(? AS DATE) AND CAST(? AS DATE)
|
||||
""");
|
||||
if (query.vin() != null) {
|
||||
where.append(" AND vin = ?\n");
|
||||
}
|
||||
if (query.eventType() != null) {
|
||||
where.append(" AND event_type = ?\n");
|
||||
}
|
||||
if (query.eventTimeFrom() != null) {
|
||||
where.append(" AND event_time_ms >= ?\n");
|
||||
}
|
||||
if (query.eventTimeTo() != null) {
|
||||
where.append(" AND event_time_ms <= ?\n");
|
||||
}
|
||||
String sql = """
|
||||
SELECT event_id, protocol, event_type, vin,
|
||||
event_time_ms, ingest_time_ms,
|
||||
raw_archive_uri, metadata_json, payload_json
|
||||
FROM event_records
|
||||
%s
|
||||
ORDER BY event_time_ms %s, ingest_time_ms %s, event_id %s
|
||||
LIMIT ?
|
||||
""".formatted(where, order, order, order);
|
||||
try (Connection connection = DriverManager.getConnection(indexJdbcUrl());
|
||||
PreparedStatement ps = connection.prepareStatement(sql)) {
|
||||
int index = 1;
|
||||
ps.setString(index++, query.protocol().name());
|
||||
ps.setString(index++, query.dateFrom().toString());
|
||||
ps.setString(index++, query.dateTo().toString());
|
||||
if (query.vin() != null) {
|
||||
ps.setString(index++, query.vin());
|
||||
}
|
||||
if (query.eventType() != null) {
|
||||
ps.setString(index++, query.eventType());
|
||||
}
|
||||
if (query.eventTimeFrom() != null) {
|
||||
ps.setLong(index++, query.eventTimeFrom().toEpochMilli());
|
||||
}
|
||||
if (query.eventTimeTo() != null) {
|
||||
ps.setLong(index++, query.eventTimeTo().toEpochMilli());
|
||||
}
|
||||
ps.setInt(index, query.limit());
|
||||
try (ResultSet rs = ps.executeQuery()) {
|
||||
List<EventFileRecord> result = new ArrayList<>();
|
||||
while (rs.next()) {
|
||||
result.add(record(rs));
|
||||
}
|
||||
return result;
|
||||
}
|
||||
} catch (SQLException e) {
|
||||
throw new IOException("query duckdb event index failed", e);
|
||||
}
|
||||
}
|
||||
|
||||
private void writePartition(Partition partition, List<EventFileRecord> records) throws IOException {
|
||||
Path dir = partition.dir(root);
|
||||
Files.createDirectories(dir);
|
||||
Path partFile = dir.resolve("events.parquet");
|
||||
Path tempFile = dir.resolve("events-" + UUID.randomUUID() + ".tmp.parquet");
|
||||
|
||||
try (Connection connection = DriverManager.getConnection("jdbc:duckdb:");
|
||||
Statement statement = connection.createStatement()) {
|
||||
statement.execute("""
|
||||
CREATE TEMPORARY TABLE event_records (
|
||||
event_id VARCHAR,
|
||||
protocol VARCHAR,
|
||||
event_type VARCHAR,
|
||||
vin VARCHAR,
|
||||
event_time_ms BIGINT,
|
||||
event_time VARCHAR,
|
||||
ingest_time_ms BIGINT,
|
||||
ingest_time VARCHAR,
|
||||
raw_archive_uri VARCHAR,
|
||||
metadata_json VARCHAR,
|
||||
payload_json VARCHAR
|
||||
)
|
||||
""");
|
||||
try (PreparedStatement ps = connection.prepareStatement("""
|
||||
INSERT INTO event_records VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""")) {
|
||||
for (EventFileRecord record : records) {
|
||||
ps.setString(1, record.eventId());
|
||||
ps.setString(2, record.protocol().name());
|
||||
ps.setString(3, record.eventType());
|
||||
ps.setString(4, record.vin());
|
||||
ps.setLong(5, record.eventTime().toEpochMilli());
|
||||
ps.setString(6, record.eventTime().toString());
|
||||
ps.setLong(7, record.ingestTime().toEpochMilli());
|
||||
ps.setString(8, record.ingestTime().toString());
|
||||
ps.setString(9, record.rawArchiveUri());
|
||||
ps.setString(10, objectMapper.writeValueAsString(record.metadata()));
|
||||
ps.setString(11, record.payloadJson());
|
||||
ps.addBatch();
|
||||
}
|
||||
ps.executeBatch();
|
||||
}
|
||||
if (Files.exists(partFile)) {
|
||||
statement.execute("""
|
||||
CREATE TEMPORARY TABLE combined_records AS
|
||||
SELECT event_id, protocol, event_type, vin,
|
||||
event_time_ms, event_time, ingest_time_ms, ingest_time,
|
||||
raw_archive_uri, metadata_json, payload_json
|
||||
FROM read_parquet('%s')
|
||||
UNION ALL
|
||||
SELECT event_id, protocol, event_type, vin,
|
||||
event_time_ms, event_time, ingest_time_ms, ingest_time,
|
||||
raw_archive_uri, metadata_json, payload_json
|
||||
FROM event_records
|
||||
""".formatted(escapeSql(partFile)));
|
||||
statement.execute("COPY combined_records TO '" + escapeSql(tempFile) + "' (FORMAT parquet)");
|
||||
} else {
|
||||
statement.execute("COPY event_records TO '" + escapeSql(tempFile) + "' (FORMAT parquet)");
|
||||
}
|
||||
Files.move(tempFile, partFile, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE);
|
||||
} catch (SQLException e) {
|
||||
throw new IOException("write parquet event store failed: " + partFile, e);
|
||||
} finally {
|
||||
Files.deleteIfExists(tempFile);
|
||||
}
|
||||
}
|
||||
|
||||
private void appendIndex(Partition partition, List<EventFileRecord> records) throws IOException {
|
||||
ensureIndexInitialized();
|
||||
try (Connection connection = DriverManager.getConnection(indexJdbcUrl())) {
|
||||
insertIndex(connection, partition, records);
|
||||
} catch (SQLException e) {
|
||||
throw new IOException("write duckdb event index failed", e);
|
||||
}
|
||||
}
|
||||
|
||||
private void ensureIndexInitialized() throws IOException {
|
||||
if (indexInitialized) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
Files.createDirectories(root);
|
||||
} catch (IOException e) {
|
||||
throw new IOException("create event store root failed: " + root, e);
|
||||
}
|
||||
try (Connection connection = DriverManager.getConnection(indexJdbcUrl());
|
||||
Statement statement = connection.createStatement()) {
|
||||
createIndexSchema(statement);
|
||||
if (isIndexEmpty(statement)) {
|
||||
backfillIndexFromParquet(statement);
|
||||
}
|
||||
indexInitialized = true;
|
||||
} catch (SQLException e) {
|
||||
throw new IOException("initialize duckdb event index failed", e);
|
||||
}
|
||||
}
|
||||
|
||||
private void createIndexSchema(Statement statement) throws SQLException {
|
||||
statement.execute("""
|
||||
CREATE TABLE IF NOT EXISTS event_records (
|
||||
event_id VARCHAR PRIMARY KEY,
|
||||
protocol VARCHAR,
|
||||
event_type VARCHAR,
|
||||
vin VARCHAR,
|
||||
event_time_ms BIGINT,
|
||||
ingest_time_ms BIGINT,
|
||||
partition_date DATE,
|
||||
raw_archive_uri VARCHAR,
|
||||
metadata_json VARCHAR,
|
||||
payload_json VARCHAR
|
||||
)
|
||||
""");
|
||||
statement.execute("CREATE INDEX IF NOT EXISTS event_records_protocol_date_time_idx ON event_records(protocol, partition_date, event_time_ms)");
|
||||
statement.execute("CREATE INDEX IF NOT EXISTS event_records_vin_date_time_idx ON event_records(vin, partition_date, event_time_ms)");
|
||||
statement.execute("CREATE INDEX IF NOT EXISTS event_records_vin_type_date_time_idx ON event_records(protocol, vin, event_type, partition_date, event_time_ms)");
|
||||
statement.execute("CREATE INDEX IF NOT EXISTS event_records_raw_archive_uri_idx ON event_records(raw_archive_uri)");
|
||||
}
|
||||
|
||||
private boolean isIndexEmpty(Statement statement) throws SQLException {
|
||||
try (ResultSet rs = statement.executeQuery("SELECT COUNT(*) FROM event_records")) {
|
||||
return rs.next() && rs.getLong(1) == 0L;
|
||||
}
|
||||
}
|
||||
|
||||
private void backfillIndexFromParquet(Statement statement) throws SQLException, IOException {
|
||||
List<Path> files = allParquetFiles();
|
||||
if (files.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
String fileList = toDuckDbList(files);
|
||||
statement.execute("""
|
||||
INSERT OR IGNORE INTO event_records
|
||||
SELECT event_id, protocol, event_type, vin,
|
||||
event_time_ms, ingest_time_ms,
|
||||
CAST(date AS DATE) AS partition_date,
|
||||
raw_archive_uri, metadata_json, payload_json
|
||||
FROM read_parquet(%s, hive_partitioning = true)
|
||||
""".formatted(fileList));
|
||||
}
|
||||
|
||||
private void insertIndex(Connection connection, Partition partition, List<EventFileRecord> records)
|
||||
throws SQLException, IOException {
|
||||
try (PreparedStatement ps = connection.prepareStatement("""
|
||||
INSERT OR REPLACE INTO event_records VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""")) {
|
||||
for (EventFileRecord record : records) {
|
||||
ps.setString(1, record.eventId());
|
||||
ps.setString(2, record.protocol().name());
|
||||
ps.setString(3, record.eventType());
|
||||
ps.setString(4, record.vin());
|
||||
ps.setLong(5, record.eventTime().toEpochMilli());
|
||||
ps.setLong(6, record.ingestTime().toEpochMilli());
|
||||
ps.setString(7, partition.date().toString());
|
||||
ps.setString(8, record.rawArchiveUri());
|
||||
ps.setString(9, objectMapper.writeValueAsString(record.metadata()));
|
||||
ps.setString(10, record.payloadJson());
|
||||
ps.addBatch();
|
||||
}
|
||||
ps.executeBatch();
|
||||
}
|
||||
}
|
||||
|
||||
private List<Path> parquetFiles(EventFileQuery query) throws IOException {
|
||||
List<Path> files = new ArrayList<>();
|
||||
LocalDate date = query.dateFrom();
|
||||
while (!date.isAfter(query.dateTo())) {
|
||||
Path dir = query.vin() == null
|
||||
? partitionDir(query.protocol(), date)
|
||||
: partitionDir(query.protocol(), date).resolve("vehicle=" + storageName(query.vin()));
|
||||
if (Files.isDirectory(dir)) {
|
||||
try (Stream<Path> stream = Files.walk(dir)) {
|
||||
stream.filter(path -> path.getFileName().toString().endsWith(".parquet"))
|
||||
.sorted()
|
||||
.forEach(files::add);
|
||||
}
|
||||
}
|
||||
date = date.plusDays(1);
|
||||
}
|
||||
return files;
|
||||
}
|
||||
|
||||
private List<Path> allParquetFiles() throws IOException {
|
||||
if (!Files.isDirectory(root)) {
|
||||
return List.of();
|
||||
}
|
||||
try (Stream<Path> stream = Files.walk(root)) {
|
||||
return stream.filter(path -> path.getFileName().toString().endsWith(".parquet"))
|
||||
.sorted()
|
||||
.toList();
|
||||
}
|
||||
}
|
||||
|
||||
private EventFileRecord record(ResultSet rs) throws SQLException, IOException {
|
||||
return new EventFileRecord(
|
||||
rs.getString("event_id"),
|
||||
protocol(rs.getString("protocol")),
|
||||
rs.getString("event_type"),
|
||||
rs.getString("vin"),
|
||||
Instant.ofEpochMilli(rs.getLong("event_time_ms")),
|
||||
Instant.ofEpochMilli(rs.getLong("ingest_time_ms")),
|
||||
rs.getString("raw_archive_uri"),
|
||||
readMetadata(rs.getString("metadata_json")),
|
||||
rs.getString("payload_json")
|
||||
);
|
||||
}
|
||||
|
||||
private Partition partition(EventFileRecord record) {
|
||||
return new Partition(
|
||||
record.protocol(),
|
||||
LocalDate.ofInstant(record.eventTime(), partitionZone),
|
||||
storageName(record.vin()));
|
||||
}
|
||||
|
||||
private Path partitionDir(ProtocolId protocol, LocalDate date) {
|
||||
return root.resolve("protocol=" + protocol.name()).resolve("date=" + date);
|
||||
}
|
||||
|
||||
private static ProtocolId protocol(String value) {
|
||||
if (value == null || value.isBlank()) {
|
||||
return ProtocolId.UNKNOWN;
|
||||
}
|
||||
try {
|
||||
return ProtocolId.valueOf(value);
|
||||
} catch (IllegalArgumentException ex) {
|
||||
return ProtocolId.UNKNOWN;
|
||||
}
|
||||
}
|
||||
|
||||
private Map<String, String> readMetadata(String json) throws IOException {
|
||||
if (json == null || json.isBlank()) {
|
||||
return Map.of();
|
||||
}
|
||||
return objectMapper.readValue(json, STRING_MAP);
|
||||
}
|
||||
|
||||
private static String toDuckDbList(List<Path> files) {
|
||||
return files.stream()
|
||||
.map(path -> "'" + escapeSql(path) + "'")
|
||||
.reduce((left, right) -> left + "," + right)
|
||||
.map(paths -> "[" + paths + "]")
|
||||
.orElse("[]");
|
||||
}
|
||||
|
||||
private static String escapeSql(Path path) {
|
||||
return escapeSql(path.toAbsolutePath().toString());
|
||||
}
|
||||
|
||||
private static String escapeSql(String value) {
|
||||
return value.replace("'", "''");
|
||||
}
|
||||
|
||||
private String indexJdbcUrl() {
|
||||
return "jdbc:duckdb:" + indexPath;
|
||||
}
|
||||
|
||||
private static String storageName(String vin) {
|
||||
if (vin == null || vin.isBlank()) {
|
||||
return "_unknown";
|
||||
}
|
||||
return vin.replaceAll("[^A-Za-z0-9._-]", "_");
|
||||
}
|
||||
|
||||
private record Partition(ProtocolId protocol, LocalDate date, String vehicle) {
|
||||
private Path dir(Path root) {
|
||||
return root.resolve("protocol=" + protocol.name())
|
||||
.resolve("date=" + date)
|
||||
.resolve("vehicle=" + vehicle)
|
||||
.resolve("header=" + HEADER_VERSION);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
package com.lingniu.ingest.eventfilestore;
|
||||
|
||||
import com.lingniu.ingest.api.ProtocolId;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.time.LocalDate;
|
||||
|
||||
public record EventFileQuery(
|
||||
ProtocolId protocol,
|
||||
LocalDate dateFrom,
|
||||
LocalDate dateTo,
|
||||
Instant eventTimeFrom,
|
||||
Instant eventTimeTo,
|
||||
Order order,
|
||||
int limit,
|
||||
String vin,
|
||||
String eventType
|
||||
) {
|
||||
public enum Order {
|
||||
ASC,
|
||||
DESC
|
||||
}
|
||||
|
||||
public EventFileQuery {
|
||||
if (protocol == null) {
|
||||
throw new IllegalArgumentException("protocol must not be null");
|
||||
}
|
||||
if (dateFrom == null || dateTo == null) {
|
||||
throw new IllegalArgumentException("date range must not be null");
|
||||
}
|
||||
if (dateTo.isBefore(dateFrom)) {
|
||||
throw new IllegalArgumentException("dateTo must not be before dateFrom");
|
||||
}
|
||||
if (eventTimeFrom != null && eventTimeTo != null && eventTimeTo.isBefore(eventTimeFrom)) {
|
||||
throw new IllegalArgumentException("eventTimeTo must not be before eventTimeFrom");
|
||||
}
|
||||
order = order == null ? Order.ASC : order;
|
||||
limit = limit <= 0 ? 100 : limit;
|
||||
vin = vin == null || vin.isBlank() ? null : vin.trim();
|
||||
eventType = eventType == null || eventType.isBlank() ? null : eventType.trim();
|
||||
}
|
||||
|
||||
public EventFileQuery(ProtocolId protocol,
|
||||
LocalDate dateFrom,
|
||||
LocalDate dateTo,
|
||||
Order order,
|
||||
int limit,
|
||||
String vin,
|
||||
String eventType) {
|
||||
this(protocol, dateFrom, dateTo, null, null, order, limit, vin, eventType);
|
||||
}
|
||||
|
||||
public EventFileQuery(ProtocolId protocol,
|
||||
LocalDate dateFrom,
|
||||
LocalDate dateTo,
|
||||
Order order,
|
||||
int limit) {
|
||||
this(protocol, dateFrom, dateTo, order, limit, null, null);
|
||||
}
|
||||
|
||||
public EventFileQuery(ProtocolId protocol,
|
||||
LocalDate dateFrom,
|
||||
LocalDate dateTo,
|
||||
Order order,
|
||||
int limit,
|
||||
String vin) {
|
||||
this(protocol, dateFrom, dateTo, order, limit, vin, null);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
package com.lingniu.ingest.eventfilestore;
|
||||
|
||||
import com.lingniu.ingest.api.ProtocolId;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* 文件型明细库的一条可查询记录。
|
||||
*
|
||||
* <p>{@code payloadJson} 保存统一 telemetry snapshot JSON；公共列只保留排序、分区、
|
||||
* 追溯和通用展示需要的最小字段。
|
||||
*/
|
||||
public record EventFileRecord(
|
||||
String eventId,
|
||||
ProtocolId protocol,
|
||||
String eventType,
|
||||
String vin,
|
||||
Instant eventTime,
|
||||
Instant ingestTime,
|
||||
String rawArchiveUri,
|
||||
Map<String, String> metadata,
|
||||
String payloadJson
|
||||
) {
|
||||
public EventFileRecord {
|
||||
if (eventId == null || eventId.isBlank()) {
|
||||
throw new IllegalArgumentException("eventId must not be blank");
|
||||
}
|
||||
if (protocol == null) {
|
||||
throw new IllegalArgumentException("protocol must not be null");
|
||||
}
|
||||
if (eventTime == null) {
|
||||
throw new IllegalArgumentException("eventTime must not be null");
|
||||
}
|
||||
if (ingestTime == null) {
|
||||
throw new IllegalArgumentException("ingestTime must not be null");
|
||||
}
|
||||
eventType = eventType == null ? "" : eventType;
|
||||
vin = vin == null ? "" : vin;
|
||||
rawArchiveUri = rawArchiveUri == null ? "" : rawArchiveUri;
|
||||
metadata = metadata == null ? Map.of() : Map.copyOf(metadata);
|
||||
payloadJson = payloadJson == null || payloadJson.isBlank() ? "{}" : payloadJson;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
package com.lingniu.ingest.eventfilestore;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.List;
|
||||
|
||||
public interface EventFileStore {
|
||||
|
||||
default void append(EventFileRecord record) throws IOException {
|
||||
appendAll(List.of(record));
|
||||
}
|
||||
|
||||
void appendAll(List<EventFileRecord> records) throws IOException;
|
||||
|
||||
List<EventFileRecord> query(EventFileQuery query) throws IOException;
|
||||
|
||||
default EventFileRecord findByRawArchiveUri(String rawArchiveUri) throws IOException {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,244 @@
|
||||
package com.lingniu.ingest.eventfilestore;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.lingniu.ingest.api.event.RawArchiveKeys;
|
||||
import com.lingniu.ingest.api.event.TelemetrySnapshot;
|
||||
import com.lingniu.ingest.api.event.VehicleEvent;
|
||||
import com.lingniu.ingest.api.event.VehicleEventTelemetrySnapshotMapper;
|
||||
import com.lingniu.ingest.api.sink.EventSink;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
* EventBus 出口：把解析后的 {@link VehicleEvent} 明细写入 Parquet 文件库。
|
||||
*
|
||||
* <p>{@link VehicleEvent.RawArchive} 只写可查询索引，不写原始 bytes；原始 bytes 本体仍由
|
||||
* sink-archive 独立冷存。
|
||||
*/
|
||||
public final class EventFileStoreSink implements EventSink, AutoCloseable {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(EventFileStoreSink.class);
|
||||
|
||||
private final EventFileStore store;
|
||||
private final ObjectMapper objectMapper;
|
||||
private final int batchSize;
|
||||
private final List<EventFileRecord> buffer = new ArrayList<>();
|
||||
private final ScheduledExecutorService flusher;
|
||||
|
||||
public EventFileStoreSink(EventFileStore store, ObjectMapper objectMapper) {
|
||||
this(store, objectMapper, 1);
|
||||
}
|
||||
|
||||
public EventFileStoreSink(EventFileStore store, ObjectMapper objectMapper, int batchSize) {
|
||||
this(store, objectMapper, batchSize, 0);
|
||||
}
|
||||
|
||||
public EventFileStoreSink(EventFileStore store,
|
||||
ObjectMapper objectMapper,
|
||||
int batchSize,
|
||||
long flushIntervalMillis) {
|
||||
if (store == null) {
|
||||
throw new IllegalArgumentException("store must not be null");
|
||||
}
|
||||
this.store = store;
|
||||
this.objectMapper = objectMapper == null ? new ObjectMapper() : objectMapper;
|
||||
this.batchSize = Math.max(1, batchSize);
|
||||
this.flusher = flushIntervalMillis > 0 && this.batchSize > 1
|
||||
? Executors.newSingleThreadScheduledExecutor(r -> {
|
||||
Thread t = Thread.ofVirtual().name("event-file-store-flush-", 0).factory().newThread(r);
|
||||
t.setDaemon(true);
|
||||
return t;
|
||||
})
|
||||
: null;
|
||||
if (flusher != null) {
|
||||
flusher.scheduleWithFixedDelay(this::flushQuietly,
|
||||
flushIntervalMillis, flushIntervalMillis, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public String name() {
|
||||
return "event-file-store";
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean accepts(VehicleEvent event) {
|
||||
return event != null
|
||||
&& !(event instanceof VehicleEvent.Realtime)
|
||||
&& !(event instanceof VehicleEvent.Location);
|
||||
}
|
||||
|
||||
@Override
|
||||
public CompletableFuture<Void> publish(VehicleEvent event) {
|
||||
if (!accepts(event)) {
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
CompletableFuture<Void> future = new CompletableFuture<>();
|
||||
try {
|
||||
EventFileRecord record = toRecord(event);
|
||||
if (batchSize <= 1) {
|
||||
store.append(record);
|
||||
} else {
|
||||
appendBuffered(record);
|
||||
}
|
||||
future.complete(null);
|
||||
} catch (Exception e) {
|
||||
future.completeExceptionally(e);
|
||||
}
|
||||
return future;
|
||||
}
|
||||
|
||||
@Override
|
||||
public CompletableFuture<Void> publishBatch(List<VehicleEvent> batch) {
|
||||
if (batch == null || batch.isEmpty()) {
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
CompletableFuture<Void> future = new CompletableFuture<>();
|
||||
try {
|
||||
List<EventFileRecord> records = new ArrayList<>();
|
||||
for (VehicleEvent event : batch) {
|
||||
if (accepts(event)) {
|
||||
records.add(toRecord(event));
|
||||
}
|
||||
}
|
||||
if (!records.isEmpty()) {
|
||||
store.appendAll(records);
|
||||
}
|
||||
future.complete(null);
|
||||
} catch (Exception e) {
|
||||
future.completeExceptionally(e);
|
||||
}
|
||||
return future;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws Exception {
|
||||
if (flusher != null) {
|
||||
flusher.shutdownNow();
|
||||
}
|
||||
flush();
|
||||
}
|
||||
|
||||
private void appendBuffered(EventFileRecord record) throws IOException {
|
||||
List<EventFileRecord> toFlush = null;
|
||||
synchronized (buffer) {
|
||||
buffer.add(record);
|
||||
if (buffer.size() >= batchSize) {
|
||||
toFlush = new ArrayList<>(buffer);
|
||||
buffer.clear();
|
||||
}
|
||||
}
|
||||
if (toFlush != null) {
|
||||
store.appendAll(toFlush);
|
||||
}
|
||||
}
|
||||
|
||||
private void flush() throws IOException {
|
||||
List<EventFileRecord> toFlush;
|
||||
synchronized (buffer) {
|
||||
if (buffer.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
toFlush = new ArrayList<>(buffer);
|
||||
buffer.clear();
|
||||
}
|
||||
store.appendAll(toFlush);
|
||||
}
|
||||
|
||||
private void flushQuietly() {
|
||||
try {
|
||||
flush();
|
||||
} catch (Exception e) {
|
||||
log.warn("event file store scheduled flush failed", e);
|
||||
}
|
||||
}
|
||||
|
||||
private EventFileRecord toRecord(VehicleEvent event) throws IOException {
|
||||
if (event instanceof VehicleEvent.RawArchive raw) {
|
||||
return rawArchiveIndexRecord(raw);
|
||||
}
|
||||
TelemetrySnapshot snapshot = VehicleEventTelemetrySnapshotMapper.toSnapshot(event)
|
||||
.orElseThrow(() -> new IllegalArgumentException("event cannot be converted to telemetry snapshot"));
|
||||
return new EventFileRecord(
|
||||
event.eventId(),
|
||||
event.source(),
|
||||
snapshot.eventType(),
|
||||
event.vin(),
|
||||
event.eventTime(),
|
||||
event.ingestTime(),
|
||||
snapshot.rawArchiveUri(),
|
||||
event.metadata(),
|
||||
objectMapper.writeValueAsString(snapshotPayload(snapshot))
|
||||
);
|
||||
}
|
||||
|
||||
private EventFileRecord rawArchiveIndexRecord(VehicleEvent.RawArchive raw) throws IOException {
|
||||
String rawArchiveKey = RawArchiveKeys.key(raw);
|
||||
String rawArchiveUri = RawArchiveKeys.logicalUri(rawArchiveKey);
|
||||
Map<String, String> metadata = new LinkedHashMap<>();
|
||||
if (raw.metadata() != null) {
|
||||
metadata.putAll(raw.metadata());
|
||||
}
|
||||
metadata.putIfAbsent(RawArchiveKeys.META_EVENT_ID, raw.eventId());
|
||||
metadata.putIfAbsent(RawArchiveKeys.META_KEY, rawArchiveKey);
|
||||
metadata.putIfAbsent(RawArchiveKeys.META_URI, rawArchiveUri);
|
||||
return new EventFileRecord(
|
||||
raw.eventId(),
|
||||
raw.source(),
|
||||
"RAW_ARCHIVE",
|
||||
raw.vin(),
|
||||
raw.eventTime(),
|
||||
raw.ingestTime(),
|
||||
rawArchiveUri,
|
||||
metadata,
|
||||
objectMapper.writeValueAsString(rawArchivePayload(raw, rawArchiveKey, rawArchiveUri))
|
||||
);
|
||||
}
|
||||
|
||||
private static Map<String, Object> snapshotPayload(TelemetrySnapshot snapshot) {
|
||||
Map<String, Object> payload = new LinkedHashMap<>();
|
||||
payload.put("eventId", snapshot.eventId());
|
||||
payload.put("vin", snapshot.vin());
|
||||
payload.put("protocol", snapshot.protocol().name());
|
||||
payload.put("eventType", snapshot.eventType());
|
||||
payload.put("eventTime", snapshot.eventTime().toString());
|
||||
payload.put("ingestTime", snapshot.ingestTime().toString());
|
||||
payload.put("rawArchiveUri", snapshot.rawArchiveUri());
|
||||
payload.put("metadata", snapshot.metadata());
|
||||
payload.put("fields", snapshot.fields());
|
||||
return payload;
|
||||
}
|
||||
|
||||
private static Map<String, Object> rawArchivePayload(VehicleEvent.RawArchive raw,
|
||||
String rawArchiveKey,
|
||||
String rawArchiveUri) {
|
||||
Map<String, Object> payload = new LinkedHashMap<>();
|
||||
payload.put("eventId", raw.eventId());
|
||||
payload.put("vin", raw.vin());
|
||||
payload.put("protocol", raw.source().name());
|
||||
payload.put("eventType", "RAW_ARCHIVE");
|
||||
payload.put("eventTime", raw.eventTime().toString());
|
||||
payload.put("ingestTime", raw.ingestTime().toString());
|
||||
payload.put("rawArchiveUri", rawArchiveUri);
|
||||
payload.put("rawArchiveKey", rawArchiveKey);
|
||||
payload.put("command", hex(raw.command()));
|
||||
payload.put("infoType", hex(raw.infoType()));
|
||||
payload.put("rawSizeBytes", raw.rawBytes() == null ? 0 : raw.rawBytes().length);
|
||||
payload.put("metadata", raw.metadata());
|
||||
return payload;
|
||||
}
|
||||
|
||||
private static String hex(int value) {
|
||||
return "0x" + String.format("%04X", value);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
package com.lingniu.ingest.eventfilestore.config;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
|
||||
import com.lingniu.ingest.eventfilestore.DuckDbParquetEventFileStore;
|
||||
import com.lingniu.ingest.eventfilestore.EventFileStore;
|
||||
import com.lingniu.ingest.eventfilestore.EventFileStoreSink;
|
||||
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.beans.factory.ObjectProvider;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
|
||||
import java.nio.file.Path;
|
||||
import java.time.ZoneId;
|
||||
|
||||
@AutoConfiguration
|
||||
@EnableConfigurationProperties(EventFileStoreProperties.class)
|
||||
@ConditionalOnProperty(
|
||||
prefix = "lingniu.ingest.event-file-store",
|
||||
name = "enabled",
|
||||
havingValue = "true")
|
||||
public class EventFileStoreAutoConfiguration {
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public EventFileStore eventFileStore(EventFileStoreProperties properties,
|
||||
ObjectProvider<ObjectMapper> objectMapper) {
|
||||
ObjectMapper mapper = mapper(objectMapper);
|
||||
return new DuckDbParquetEventFileStore(
|
||||
Path.of(properties.getPath()),
|
||||
ZoneId.of(properties.getZoneId()),
|
||||
mapper);
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public EventFileStoreSink eventFileStoreSink(EventFileStore store,
|
||||
EventFileStoreProperties properties,
|
||||
ObjectProvider<ObjectMapper> objectMapper) {
|
||||
return new EventFileStoreSink(
|
||||
store,
|
||||
mapper(objectMapper),
|
||||
properties.getBatchSize(),
|
||||
properties.getFlushIntervalMillis());
|
||||
}
|
||||
|
||||
private static ObjectMapper mapper(ObjectProvider<ObjectMapper> provider) {
|
||||
ObjectMapper base = provider.getIfAvailable(ObjectMapper::new);
|
||||
return base.copy().registerModule(new JavaTimeModule());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,72 @@
|
||||
package com.lingniu.ingest.eventfilestore.config;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
@ConfigurationProperties(prefix = "lingniu.ingest.event-file-store")
|
||||
public class EventFileStoreProperties {
|
||||
|
||||
/**
|
||||
* 默认关闭，避免仅引入依赖就开始写明细文件。
|
||||
*/
|
||||
private boolean enabled = false;
|
||||
|
||||
/**
|
||||
* Parquet 文件库根路径。
|
||||
*/
|
||||
private String path = "./event-store";
|
||||
|
||||
/**
|
||||
* 按事件时间分区时使用的业务时区。
|
||||
*/
|
||||
private String zoneId = "Asia/Shanghai";
|
||||
|
||||
/**
|
||||
* Sink 缓冲多少条事件后批量写一个 Parquet part 文件。
|
||||
*/
|
||||
private int batchSize = 500;
|
||||
|
||||
/**
|
||||
* 未达到 batchSize 时的最长刷盘间隔。
|
||||
*/
|
||||
private long flushIntervalMillis = 1000;
|
||||
|
||||
public boolean isEnabled() {
|
||||
return enabled;
|
||||
}
|
||||
|
||||
public void setEnabled(boolean enabled) {
|
||||
this.enabled = enabled;
|
||||
}
|
||||
|
||||
public String getPath() {
|
||||
return path;
|
||||
}
|
||||
|
||||
public void setPath(String path) {
|
||||
this.path = path;
|
||||
}
|
||||
|
||||
public String getZoneId() {
|
||||
return zoneId;
|
||||
}
|
||||
|
||||
public void setZoneId(String zoneId) {
|
||||
this.zoneId = zoneId;
|
||||
}
|
||||
|
||||
public int getBatchSize() {
|
||||
return batchSize;
|
||||
}
|
||||
|
||||
public void setBatchSize(int batchSize) {
|
||||
this.batchSize = batchSize;
|
||||
}
|
||||
|
||||
public long getFlushIntervalMillis() {
|
||||
return flushIntervalMillis;
|
||||
}
|
||||
|
||||
public void setFlushIntervalMillis(long flushIntervalMillis) {
|
||||
this.flushIntervalMillis = flushIntervalMillis;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
com.lingniu.ingest.eventfilestore.config.EventFileStoreAutoConfiguration
|
||||
@@ -0,0 +1,257 @@
|
||||
package com.lingniu.ingest.eventfilestore;
|
||||
|
||||
import com.lingniu.ingest.api.ProtocolId;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.io.TempDir;
|
||||
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.time.Instant;
|
||||
import java.time.LocalDate;
|
||||
import java.time.ZoneId;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
class DuckDbParquetEventFileStoreTest {
|
||||
|
||||
@TempDir
|
||||
Path tempDir;
|
||||
|
||||
@Test
|
||||
void appendsRecordsToParquetAndQueriesByTimeAscendingAndDescending() throws Exception {
|
||||
EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"));
|
||||
EventFileRecord first = record("e1", Instant.parse("2026-06-21T16:00:02Z"), "粤A00001");
|
||||
EventFileRecord second = record("e2", Instant.parse("2026-06-21T16:00:01Z"), "粤A00002");
|
||||
|
||||
store.appendAll(List.of(first, second));
|
||||
|
||||
Path dateDir = tempDir.resolve("protocol=GB32960").resolve("date=2026-06-22");
|
||||
assertThat(parquetFiles(dateDir)).hasSize(2);
|
||||
|
||||
EventFileQuery ascending = new EventFileQuery(
|
||||
ProtocolId.GB32960,
|
||||
LocalDate.parse("2026-06-22"),
|
||||
LocalDate.parse("2026-06-22"),
|
||||
EventFileQuery.Order.ASC,
|
||||
100);
|
||||
assertThat(store.query(ascending))
|
||||
.extracting(EventFileRecord::eventId)
|
||||
.containsExactly("e2", "e1");
|
||||
|
||||
EventFileQuery descending = new EventFileQuery(
|
||||
ProtocolId.GB32960,
|
||||
LocalDate.parse("2026-06-22"),
|
||||
LocalDate.parse("2026-06-22"),
|
||||
EventFileQuery.Order.DESC,
|
||||
100);
|
||||
assertThat(store.query(descending))
|
||||
.extracting(EventFileRecord::eventId)
|
||||
.containsExactly("e1", "e2");
|
||||
}
|
||||
|
||||
@Test
|
||||
void queriesAcrossMultipleDaysInTimeOrder() throws Exception {
|
||||
EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"));
|
||||
store.append(record("d1", Instant.parse("2026-06-20T16:00:00Z"), "粤A00001"));
|
||||
store.append(record("d2", Instant.parse("2026-06-21T16:00:00Z"), "粤A00001"));
|
||||
|
||||
EventFileQuery query = new EventFileQuery(
|
||||
ProtocolId.GB32960,
|
||||
LocalDate.parse("2026-06-21"),
|
||||
LocalDate.parse("2026-06-22"),
|
||||
EventFileQuery.Order.ASC,
|
||||
100);
|
||||
|
||||
assertThat(store.query(query))
|
||||
.extracting(EventFileRecord::eventId)
|
||||
.containsExactly("d1", "d2");
|
||||
}
|
||||
|
||||
@Test
|
||||
void filtersByVinBeforeApplyingLimit() throws Exception {
|
||||
EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"));
|
||||
store.append(record("other-1", Instant.parse("2026-06-21T16:00:03Z"), "VIN-OTHER"));
|
||||
store.append(record("target-1", Instant.parse("2026-06-21T16:00:02Z"), "VIN-TARGET"));
|
||||
store.append(record("other-2", Instant.parse("2026-06-21T16:00:01Z"), "VIN-OTHER"));
|
||||
store.append(record("target-2", Instant.parse("2026-06-21T16:00:00Z"), "VIN-TARGET"));
|
||||
|
||||
EventFileQuery query = new EventFileQuery(
|
||||
ProtocolId.GB32960,
|
||||
LocalDate.parse("2026-06-22"),
|
||||
LocalDate.parse("2026-06-22"),
|
||||
EventFileQuery.Order.DESC,
|
||||
1,
|
||||
"VIN-TARGET");
|
||||
|
||||
assertThat(store.query(query))
|
||||
.extracting(EventFileRecord::eventId)
|
||||
.containsExactly("target-1");
|
||||
}
|
||||
|
||||
@Test
|
||||
void filtersByVinAndEventTypeBeforeApplyingLimit() throws Exception {
|
||||
EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"));
|
||||
store.append(rawRecord("raw-other", Instant.parse("2026-06-21T16:00:04Z"), "VIN-OTHER"));
|
||||
store.append(record("business-target", Instant.parse("2026-06-21T16:00:03Z"), "VIN-TARGET"));
|
||||
store.append(rawRecord("raw-target", Instant.parse("2026-06-21T16:00:02Z"), "VIN-TARGET"));
|
||||
|
||||
EventFileQuery query = new EventFileQuery(
|
||||
ProtocolId.GB32960,
|
||||
LocalDate.parse("2026-06-22"),
|
||||
LocalDate.parse("2026-06-22"),
|
||||
EventFileQuery.Order.DESC,
|
||||
1,
|
||||
"VIN-TARGET",
|
||||
"RAW_ARCHIVE");
|
||||
|
||||
assertThat(store.query(query))
|
||||
.extracting(EventFileRecord::eventId)
|
||||
.containsExactly("raw-target");
|
||||
}
|
||||
|
||||
@Test
|
||||
void filtersByExactEventTimeRangeBeforeApplyingLimit() throws Exception {
|
||||
EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"));
|
||||
store.append(rawRecord("before", Instant.parse("2026-06-21T16:00:00Z"), "VIN-TIME"));
|
||||
store.append(rawRecord("inside", Instant.parse("2026-06-21T16:00:02Z"), "VIN-TIME"));
|
||||
store.append(rawRecord("after", Instant.parse("2026-06-21T16:00:04Z"), "VIN-TIME"));
|
||||
|
||||
EventFileQuery query = new EventFileQuery(
|
||||
ProtocolId.GB32960,
|
||||
LocalDate.parse("2026-06-22"),
|
||||
LocalDate.parse("2026-06-22"),
|
||||
Instant.parse("2026-06-21T16:00:01Z"),
|
||||
Instant.parse("2026-06-21T16:00:03Z"),
|
||||
EventFileQuery.Order.ASC,
|
||||
10,
|
||||
"VIN-TIME",
|
||||
"RAW_ARCHIVE");
|
||||
|
||||
assertThat(store.query(query))
|
||||
.extracting(EventFileRecord::eventId)
|
||||
.containsExactly("inside");
|
||||
}
|
||||
|
||||
@Test
|
||||
void appendsSameVehicleDayHeaderToOneParquetFile() throws Exception {
|
||||
EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"));
|
||||
store.append(record("first", Instant.parse("2026-06-21T16:00:03Z"), "VIN-SAME"));
|
||||
store.append(record("second", Instant.parse("2026-06-21T16:00:04Z"), "VIN-SAME"));
|
||||
|
||||
Path vehicleDayDir = tempDir.resolve("protocol=GB32960")
|
||||
.resolve("date=2026-06-22")
|
||||
.resolve("vehicle=VIN-SAME")
|
||||
.resolve("header=event-records-v1");
|
||||
assertThat(parquetFiles(vehicleDayDir)).containsExactly(vehicleDayDir.resolve("events.parquet"));
|
||||
|
||||
EventFileQuery query = new EventFileQuery(
|
||||
ProtocolId.GB32960,
|
||||
LocalDate.parse("2026-06-22"),
|
||||
LocalDate.parse("2026-06-22"),
|
||||
EventFileQuery.Order.ASC,
|
||||
10,
|
||||
"VIN-SAME");
|
||||
|
||||
assertThat(store.query(query))
|
||||
.extracting(EventFileRecord::eventId)
|
||||
.containsExactly("first", "second");
|
||||
}
|
||||
|
||||
@Test
|
||||
void queriesUseSidecarIndexInsteadOfScanningParquetPartsEveryTime() throws Exception {
|
||||
EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"));
|
||||
store.append(record("indexed", Instant.parse("2026-06-21T16:00:03Z"), "VIN-INDEXED"));
|
||||
|
||||
Path vehicleDayDir = tempDir.resolve("protocol=GB32960")
|
||||
.resolve("date=2026-06-22")
|
||||
.resolve("vehicle=VIN-INDEXED")
|
||||
.resolve("header=event-records-v1");
|
||||
try (var files = Files.list(vehicleDayDir)) {
|
||||
for (Path file : files.filter(p -> p.getFileName().toString().endsWith(".parquet")).toList()) {
|
||||
Files.delete(file);
|
||||
}
|
||||
}
|
||||
|
||||
EventFileQuery query = new EventFileQuery(
|
||||
ProtocolId.GB32960,
|
||||
LocalDate.parse("2026-06-22"),
|
||||
LocalDate.parse("2026-06-22"),
|
||||
EventFileQuery.Order.DESC,
|
||||
10);
|
||||
|
||||
assertThat(store.query(query))
|
||||
.extracting(EventFileRecord::eventId)
|
||||
.containsExactly("indexed");
|
||||
}
|
||||
|
||||
@Test
|
||||
void vinQueriesReadVehicleDayParquetWithoutGlobalDuckDbIndex() throws Exception {
|
||||
EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"));
|
||||
store.append(rawRecord("raw-target", Instant.parse("2026-06-21T16:00:03Z"), "VIN-DIRECT"));
|
||||
Files.deleteIfExists(tempDir.resolve("events.duckdb"));
|
||||
|
||||
EventFileQuery query = new EventFileQuery(
|
||||
ProtocolId.GB32960,
|
||||
LocalDate.parse("2026-06-22"),
|
||||
LocalDate.parse("2026-06-22"),
|
||||
EventFileQuery.Order.DESC,
|
||||
10,
|
||||
"VIN-DIRECT",
|
||||
"RAW_ARCHIVE");
|
||||
|
||||
assertThat(store.query(query))
|
||||
.extracting(EventFileRecord::eventId)
|
||||
.containsExactly("raw-target");
|
||||
}
|
||||
|
||||
@Test
|
||||
void findsRecordByRawArchiveUriThroughSidecarIndex() throws Exception {
|
||||
EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"));
|
||||
EventFileRecord record = record("raw-001", Instant.parse("2026-06-21T16:00:03Z"), "VIN-RAW");
|
||||
store.append(record);
|
||||
|
||||
EventFileRecord found = store.findByRawArchiveUri(record.rawArchiveUri());
|
||||
|
||||
assertThat(found).isNotNull();
|
||||
assertThat(found.eventId()).isEqualTo("raw-001");
|
||||
assertThat(found.rawArchiveUri()).isEqualTo(record.rawArchiveUri());
|
||||
}
|
||||
|
||||
private static List<Path> parquetFiles(Path root) throws Exception {
|
||||
try (var files = Files.walk(root)) {
|
||||
return files.filter(p -> p.getFileName().toString().endsWith(".parquet"))
|
||||
.sorted()
|
||||
.toList();
|
||||
}
|
||||
}
|
||||
|
||||
private static EventFileRecord record(String eventId, Instant eventTime, String vin) {
|
||||
return new EventFileRecord(
|
||||
eventId,
|
||||
ProtocolId.GB32960,
|
||||
"REALTIME",
|
||||
vin,
|
||||
eventTime,
|
||||
eventTime.plusMillis(200),
|
||||
"file:///archive/" + eventId + ".bin",
|
||||
Map.of("source", "test"),
|
||||
"""
|
||||
{"gb32960":{"vehicle":{"speedKmh":42.1},"fuelCell":{"hydrogenRemainingKg":8.3}}}
|
||||
""".strip());
|
||||
}
|
||||
|
||||
private static EventFileRecord rawRecord(String eventId, Instant eventTime, String vin) {
|
||||
return new EventFileRecord(
|
||||
eventId,
|
||||
ProtocolId.GB32960,
|
||||
"RAW_ARCHIVE",
|
||||
vin,
|
||||
eventTime,
|
||||
eventTime.plusMillis(200),
|
||||
"file:///archive/" + eventId + ".bin",
|
||||
Map.of("source", "test"),
|
||||
"{}");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,154 @@
|
||||
package com.lingniu.ingest.eventfilestore;
|
||||
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.lingniu.ingest.api.ProtocolId;
|
||||
import com.lingniu.ingest.api.event.LocationPayload;
|
||||
import com.lingniu.ingest.api.event.RawArchiveKeys;
|
||||
import com.lingniu.ingest.api.event.RealtimePayload;
|
||||
import com.lingniu.ingest.api.event.VehicleEvent;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.io.TempDir;
|
||||
|
||||
import java.nio.file.Path;
|
||||
import java.time.Instant;
|
||||
import java.time.LocalDate;
|
||||
import java.time.ZoneId;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
class EventFileStoreSinkTest {
|
||||
|
||||
@TempDir
|
||||
Path tempDir;
|
||||
|
||||
private final ObjectMapper objectMapper = new ObjectMapper();
|
||||
|
||||
@Test
|
||||
void eventSinkSkipsRealtimeAndLocationEvents() throws Exception {
|
||||
EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"), objectMapper);
|
||||
EventFileStoreSink sink = new EventFileStoreSink(store, objectMapper);
|
||||
|
||||
sink.publish(realtimeEvent("evt-1")).join();
|
||||
sink.publish(locationEvent("loc-1")).join();
|
||||
|
||||
var records = store.query(new EventFileQuery(
|
||||
ProtocolId.GB32960,
|
||||
LocalDate.parse("2026-06-22"),
|
||||
LocalDate.parse("2026-06-22"),
|
||||
EventFileQuery.Order.ASC,
|
||||
10));
|
||||
|
||||
assertThat(records).isEmpty();
|
||||
}
|
||||
|
||||
@Test
|
||||
void batchedSinkFlushesRecordsWhenClosed() throws Exception {
|
||||
EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"), objectMapper);
|
||||
EventFileStoreSink sink = new EventFileStoreSink(store, objectMapper, 10);
|
||||
|
||||
sink.publish(rawEvent("raw-batch-1")).join();
|
||||
assertThat(store.query(new EventFileQuery(
|
||||
ProtocolId.GB32960,
|
||||
LocalDate.parse("2026-06-22"),
|
||||
LocalDate.parse("2026-06-22"),
|
||||
EventFileQuery.Order.ASC,
|
||||
10))).isEmpty();
|
||||
|
||||
sink.close();
|
||||
|
||||
assertThat(store.query(new EventFileQuery(
|
||||
ProtocolId.GB32960,
|
||||
LocalDate.parse("2026-06-22"),
|
||||
LocalDate.parse("2026-06-22"),
|
||||
EventFileQuery.Order.ASC,
|
||||
10)))
|
||||
.extracting(EventFileRecord::eventId)
|
||||
.containsExactly("raw-batch-1");
|
||||
}
|
||||
|
||||
@Test
|
||||
void eventSinkPersistsRawArchiveAsQueryableIndexWithoutEmbeddingRawBytes() throws Exception {
|
||||
EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"), objectMapper);
|
||||
EventFileStoreSink sink = new EventFileStoreSink(store, objectMapper);
|
||||
VehicleEvent.RawArchive raw = new VehicleEvent.RawArchive(
|
||||
"raw-1",
|
||||
"LTEST000000000808",
|
||||
ProtocolId.JT808,
|
||||
Instant.parse("2026-06-21T16:00:03Z"),
|
||||
Instant.parse("2026-06-21T16:00:04Z"),
|
||||
"trace-raw-1",
|
||||
Map.of("phone", "013800138000", "identityResolved", "true"),
|
||||
0x0200,
|
||||
0,
|
||||
new byte[]{0x7e, 0x02, 0x00, 0x7e});
|
||||
|
||||
sink.publish(raw).join();
|
||||
|
||||
var records = store.query(new EventFileQuery(
|
||||
ProtocolId.JT808,
|
||||
LocalDate.parse("2026-06-22"),
|
||||
LocalDate.parse("2026-06-22"),
|
||||
EventFileQuery.Order.ASC,
|
||||
10));
|
||||
|
||||
assertThat(records).hasSize(1);
|
||||
EventFileRecord record = records.getFirst();
|
||||
String rawArchiveKey = RawArchiveKeys.key(raw);
|
||||
assertThat(record.eventId()).isEqualTo("raw-1");
|
||||
assertThat(record.eventType()).isEqualTo("RAW_ARCHIVE");
|
||||
assertThat(record.rawArchiveUri()).isEqualTo(RawArchiveKeys.logicalUri(rawArchiveKey));
|
||||
assertThat(record.metadata())
|
||||
.containsEntry(RawArchiveKeys.META_KEY, rawArchiveKey)
|
||||
.containsEntry(RawArchiveKeys.META_URI, RawArchiveKeys.logicalUri(rawArchiveKey));
|
||||
JsonNode payload = objectMapper.readTree(record.payloadJson());
|
||||
assertThat(payload.get("rawSizeBytes").asLong()).isEqualTo(4);
|
||||
assertThat(payload.get("command").asText()).isEqualTo("0x0200");
|
||||
assertThat(payload.toString()).doesNotContain("7e02007e");
|
||||
}
|
||||
|
||||
private static VehicleEvent.Realtime realtimeEvent(String eventId) {
|
||||
return new VehicleEvent.Realtime(
|
||||
eventId,
|
||||
"LTEST000000000001",
|
||||
ProtocolId.GB32960,
|
||||
Instant.parse("2026-06-21T16:00:00Z"),
|
||||
Instant.parse("2026-06-21T16:00:01Z"),
|
||||
"trace-1",
|
||||
Map.of("rawArchiveUri", "file:///archive/" + eventId + ".bin"),
|
||||
new RealtimePayload(
|
||||
42.1, 1234.5, 87.0, null, null,
|
||||
null, null, null, 8.3, null, null,
|
||||
RealtimePayload.VehicleState.STARTED,
|
||||
null, null, null, null, null,
|
||||
113.12, 23.45, null, null, null, null
|
||||
));
|
||||
}
|
||||
|
||||
private static VehicleEvent.Location locationEvent(String eventId) {
|
||||
return new VehicleEvent.Location(
|
||||
eventId,
|
||||
"LTEST000000000001",
|
||||
ProtocolId.GB32960,
|
||||
Instant.parse("2026-06-21T16:00:00Z"),
|
||||
Instant.parse("2026-06-21T16:00:01Z"),
|
||||
"trace-location",
|
||||
Map.of("rawArchiveUri", "file:///archive/" + eventId + ".bin"),
|
||||
new LocationPayload(113.12, 23.45, 0, 42.1, 90, 0, 0));
|
||||
}
|
||||
|
||||
private static VehicleEvent.RawArchive rawEvent(String eventId) {
|
||||
return new VehicleEvent.RawArchive(
|
||||
eventId,
|
||||
"LTEST000000000001",
|
||||
ProtocolId.GB32960,
|
||||
Instant.parse("2026-06-21T16:00:00Z"),
|
||||
Instant.parse("2026-06-21T16:00:01Z"),
|
||||
"trace-raw",
|
||||
Map.of(),
|
||||
0x0002,
|
||||
0,
|
||||
new byte[]{0x23, 0x23});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
package com.lingniu.ingest.eventfilestore.config;
|
||||
|
||||
import com.lingniu.ingest.eventfilestore.EventFileStore;
|
||||
import com.lingniu.ingest.eventfilestore.EventFileStoreSink;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigurations;
|
||||
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
|
||||
|
||||
import java.nio.file.Path;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
class EventFileStoreAutoConfigurationTest {
|
||||
|
||||
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
|
||||
.withConfiguration(AutoConfigurations.of(EventFileStoreAutoConfiguration.class));
|
||||
|
||||
@Test
|
||||
void createsStoreAndSinkWhenEnabled() {
|
||||
contextRunner
|
||||
.withPropertyValues(
|
||||
"lingniu.ingest.event-file-store.enabled=true",
|
||||
"lingniu.ingest.event-file-store.path=target/test-event-store")
|
||||
.run(context -> {
|
||||
assertThat(context).hasSingleBean(EventFileStore.class);
|
||||
assertThat(context).hasSingleBean(EventFileStoreSink.class);
|
||||
assertThat(context.getBean(EventFileStoreProperties.class).getPath())
|
||||
.isEqualTo(Path.of("target/test-event-store").toString());
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void backsOffWhenDisabled() {
|
||||
contextRunner
|
||||
.withPropertyValues("lingniu.ingest.event-file-store.enabled=false")
|
||||
.run(context -> {
|
||||
assertThat(context).doesNotHaveBean(EventFileStore.class);
|
||||
assertThat(context).doesNotHaveBean(EventFileStoreSink.class);
|
||||
});
|
||||
}
|
||||
}
|
||||
39
modules/sinks/sink-archive/pom.xml
Normal file
39
modules/sinks/sink-archive/pom.xml
Normal file
@@ -0,0 +1,39 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>com.lingniu.ingest</groupId>
|
||||
<artifactId>lingniu-vehicle-ingest</artifactId>
|
||||
<version>0.1.0-SNAPSHOT</version>
|
||||
<relativePath>../../../pom.xml</relativePath>
|
||||
</parent>
|
||||
<artifactId>sink-archive</artifactId>
|
||||
<name>sink-archive</name>
|
||||
<description>原始报文冷存：本地文件系统实现 + S3/OSS 扩展点。</description>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>com.lingniu.ingest</groupId>
|
||||
<artifactId>ingest-api</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-configuration-processor</artifactId>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.junit.jupiter</groupId>
|
||||
<artifactId>junit-jupiter</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.assertj</groupId>
|
||||
<artifactId>assertj-core</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
@@ -0,0 +1 @@
|
||||
com.lingniu.ingest.sink.archive.config.SinkArchiveAutoConfiguration
|
||||
97
modules/sinks/sink-mq/pom.xml
Normal file
97
modules/sinks/sink-mq/pom.xml
Normal file
@@ -0,0 +1,97 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>com.lingniu.ingest</groupId>
|
||||
<artifactId>lingniu-vehicle-ingest</artifactId>
|
||||
<version>0.1.0-SNAPSHOT</version>
|
||||
<relativePath>../../../pom.xml</relativePath>
|
||||
</parent>
|
||||
<artifactId>sink-mq</artifactId>
|
||||
<name>sink-mq</name>
|
||||
<description>Kafka producer + Protobuf Envelope + 重试/熔断/DLQ。</description>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>com.lingniu.ingest</groupId>
|
||||
<artifactId>ingest-api</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-configuration-processor</artifactId>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.kafka</groupId>
|
||||
<artifactId>kafka-clients</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.google.protobuf</groupId>
|
||||
<artifactId>protobuf-java</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.google.protobuf</groupId>
|
||||
<artifactId>protobuf-java-util</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.github.resilience4j</groupId>
|
||||
<artifactId>resilience4j-retry</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.github.resilience4j</groupId>
|
||||
<artifactId>resilience4j-circuitbreaker</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>kafka</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.junit.jupiter</groupId>
|
||||
<artifactId>junit-jupiter</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.assertj</groupId>
|
||||
<artifactId>assertj-core</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<extensions>
|
||||
<extension>
|
||||
<groupId>kr.motd.maven</groupId>
|
||||
<artifactId>os-maven-plugin</artifactId>
|
||||
<version>1.7.1</version>
|
||||
</extension>
|
||||
</extensions>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.xolstice.maven.plugins</groupId>
|
||||
<artifactId>protobuf-maven-plugin</artifactId>
|
||||
<version>0.6.1</version>
|
||||
<configuration>
|
||||
<protocArtifact>com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}</protocArtifact>
|
||||
</configuration>
|
||||
<executions>
|
||||
<execution>
|
||||
<goals>
|
||||
<goal>compile</goal>
|
||||
<goal>test-compile</goal>
|
||||
</goals>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</project>
|
||||
@@ -0,0 +1,167 @@
|
||||
package com.lingniu.ingest.sink.mq;
|
||||
|
||||
import com.lingniu.ingest.api.event.AlarmPayload;
|
||||
import com.lingniu.ingest.api.event.LocationPayload;
|
||||
import com.lingniu.ingest.api.event.RawArchiveKeys;
|
||||
import com.lingniu.ingest.api.event.RealtimePayload;
|
||||
import com.lingniu.ingest.api.event.TelemetryFieldValue;
|
||||
import com.lingniu.ingest.api.event.TelemetrySnapshot;
|
||||
import com.lingniu.ingest.api.event.VehicleEvent;
|
||||
import com.lingniu.ingest.api.event.VehicleEventTelemetrySnapshotMapper;
|
||||
import com.lingniu.ingest.sink.mq.proto.TelemetryField;
|
||||
import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot.Builder;
|
||||
import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
|
||||
|
||||
/**
|
||||
* 领域事件 → Protobuf Envelope 的纯函数映射。通过 switch 模式匹配穷尽处理，
|
||||
* 新增 {@link VehicleEvent} 子类型时编译期就会提示补齐分支。
|
||||
*/
|
||||
public final class EnvelopeMapper {
|
||||
|
||||
private static final String SCHEMA_VERSION = "1.0";
|
||||
private final String nodeId;
|
||||
|
||||
public EnvelopeMapper(String nodeId) {
|
||||
this.nodeId = nodeId;
|
||||
}
|
||||
|
||||
public VehicleEnvelope toEnvelope(VehicleEvent event) {
|
||||
VehicleEnvelope.Builder b = VehicleEnvelope.newBuilder()
|
||||
.setSchemaVersion(SCHEMA_VERSION)
|
||||
.setEventId(event.eventId())
|
||||
.setTraceId(event.traceId() == null ? "" : event.traceId())
|
||||
.setVin(event.vin())
|
||||
.setSource(event.source().name())
|
||||
.setEventTimeMs(event.eventTime().toEpochMilli())
|
||||
.setIngestTimeMs(event.ingestTime().toEpochMilli())
|
||||
.setIngestNodeId(nodeId);
|
||||
if (event.metadata() != null) b.putAllMetadata(event.metadata());
|
||||
VehicleEventTelemetrySnapshotMapper.toSnapshot(event)
|
||||
.map(EnvelopeMapper::buildTelemetrySnapshot)
|
||||
.ifPresent(b::setTelemetrySnapshot);
|
||||
|
||||
switch (event) {
|
||||
case VehicleEvent.Realtime r -> b.setRealtime(buildRealtime(r.payload()));
|
||||
case VehicleEvent.Location l -> b.setLocation(buildLocation(l.payload()));
|
||||
case VehicleEvent.Alarm a -> b.setAlarm(buildAlarm(a.payload()));
|
||||
case VehicleEvent.Login lg -> b.setLogin(
|
||||
com.lingniu.ingest.sink.mq.proto.LoginPayload.newBuilder()
|
||||
.setIccid(nullToEmpty(lg.iccid()))
|
||||
.setProtocolVersion(nullToEmpty(lg.protocolVersion()))
|
||||
.build());
|
||||
case VehicleEvent.Logout ignored -> b.setLogout(
|
||||
com.lingniu.ingest.sink.mq.proto.LogoutPayload.getDefaultInstance());
|
||||
case VehicleEvent.Heartbeat ignored -> b.setHeartbeat(
|
||||
com.lingniu.ingest.sink.mq.proto.HeartbeatPayload.getDefaultInstance());
|
||||
case VehicleEvent.MediaMeta m -> b.setMediaMeta(
|
||||
com.lingniu.ingest.sink.mq.proto.MediaMetaPayload.newBuilder()
|
||||
.setMediaId(nullToEmpty(m.mediaId()))
|
||||
.setMediaType(nullToEmpty(m.mediaType()))
|
||||
.setSizeBytes(m.sizeBytes())
|
||||
.setArchiveRef(nullToEmpty(m.archiveRef()))
|
||||
.build());
|
||||
case VehicleEvent.Passthrough p -> b.setPassthrough(
|
||||
com.lingniu.ingest.sink.mq.proto.PassthroughPayload.newBuilder()
|
||||
.setPassthroughType(p.passthroughType())
|
||||
.setData(com.google.protobuf.ByteString.copyFrom(
|
||||
p.data() == null ? new byte[0] : p.data()))
|
||||
.build());
|
||||
case VehicleEvent.RawArchive ra -> {
|
||||
int size = ra.rawBytes() == null ? 0 : ra.rawBytes().length;
|
||||
String key = rawArchiveKey(ra);
|
||||
String uri = rawArchiveUri(ra, key);
|
||||
b.putMetadata(RawArchiveKeys.META_KEY, key);
|
||||
b.putMetadata(RawArchiveKeys.META_URI, uri);
|
||||
b.putMetadata(RawArchiveKeys.META_EVENT_ID, ra.eventId());
|
||||
b.setRawArchive(com.lingniu.ingest.sink.mq.proto.RawArchiveRef.newBuilder()
|
||||
.setUri(uri)
|
||||
.setSizeBytes(size)
|
||||
.build());
|
||||
}
|
||||
}
|
||||
return b.build();
|
||||
}
|
||||
|
||||
private static com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot buildTelemetrySnapshot(TelemetrySnapshot snapshot) {
|
||||
Builder b = com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot.newBuilder()
|
||||
.setEventType(snapshot.eventType())
|
||||
.setRawArchiveUri(snapshot.rawArchiveUri());
|
||||
for (TelemetryFieldValue field : snapshot.fields()) {
|
||||
b.addFields(TelemetryField.newBuilder()
|
||||
.setKey(field.key())
|
||||
.setValueType(field.valueType().name())
|
||||
.setValue(field.value())
|
||||
.setUnit(field.unit())
|
||||
.setQuality(field.quality().name())
|
||||
.setSourcePath(field.sourcePath())
|
||||
.build());
|
||||
}
|
||||
return b.build();
|
||||
}
|
||||
|
||||
private static com.lingniu.ingest.sink.mq.proto.RealtimePayload buildRealtime(RealtimePayload p) {
|
||||
var b = com.lingniu.ingest.sink.mq.proto.RealtimePayload.newBuilder();
|
||||
if (p.speedKmh() != null) b.setSpeedKmh(p.speedKmh());
|
||||
if (p.totalMileageKm() != null) b.setTotalMileageKm(p.totalMileageKm());
|
||||
if (p.batterySoc() != null) b.setBatterySoc(p.batterySoc());
|
||||
if (p.batteryVoltageV() != null) b.setBatteryVoltageV(p.batteryVoltageV());
|
||||
if (p.batteryCurrentA() != null) b.setBatteryCurrentA(p.batteryCurrentA());
|
||||
if (p.fcVoltageV() != null) b.setFcVoltageV(p.fcVoltageV());
|
||||
if (p.fcCurrentA() != null) b.setFcCurrentA(p.fcCurrentA());
|
||||
if (p.fcTempC() != null) b.setFcTempC(p.fcTempC());
|
||||
if (p.hydrogenRemainingKg() != null) b.setHydrogenRemainingKg(p.hydrogenRemainingKg());
|
||||
if (p.hydrogenHighPressureMpa() != null) b.setHydrogenHighPressureMpa(p.hydrogenHighPressureMpa());
|
||||
if (p.hydrogenLowPressureMpa() != null) b.setHydrogenLowPressureMpa(p.hydrogenLowPressureMpa());
|
||||
if (p.vehicleState() != null) b.setVehicleState(p.vehicleState().name());
|
||||
if (p.chargingState() != null) b.setChargingState(p.chargingState().name());
|
||||
if (p.runningMode() != null) b.setRunningMode(p.runningMode().name());
|
||||
if (p.gearLevel() != null) b.setGearLevel(p.gearLevel());
|
||||
if (p.acceleratorPedal() != null) b.setAcceleratorPedal(p.acceleratorPedal());
|
||||
if (p.brakePedal() != null) b.setBrakePedal(p.brakePedal());
|
||||
if (p.longitude() != null) b.setLongitude(p.longitude());
|
||||
if (p.latitude() != null) b.setLatitude(p.latitude());
|
||||
if (p.altitudeM() != null) b.setAltitudeM(p.altitudeM());
|
||||
if (p.directionDeg() != null) b.setDirectionDeg(p.directionDeg());
|
||||
if (p.ambientTempC() != null) b.setAmbientTempC(p.ambientTempC());
|
||||
if (p.coolantTempC() != null) b.setCoolantTempC(p.coolantTempC());
|
||||
return b.build();
|
||||
}
|
||||
|
||||
private static com.lingniu.ingest.sink.mq.proto.LocationPayload buildLocation(LocationPayload p) {
|
||||
return com.lingniu.ingest.sink.mq.proto.LocationPayload.newBuilder()
|
||||
.setLongitude(p.longitude())
|
||||
.setLatitude(p.latitude())
|
||||
.setAltitudeM(p.altitudeM())
|
||||
.setSpeedKmh(p.speedKmh())
|
||||
.setDirectionDeg(p.directionDeg())
|
||||
.setAlarmFlag(p.alarmFlag())
|
||||
.setStatusFlag(p.statusFlag())
|
||||
.build();
|
||||
}
|
||||
|
||||
private static com.lingniu.ingest.sink.mq.proto.AlarmPayload buildAlarm(AlarmPayload p) {
|
||||
var b = com.lingniu.ingest.sink.mq.proto.AlarmPayload.newBuilder()
|
||||
.setLevel(p.level().name())
|
||||
.setAlarmTypeCode(p.alarmTypeCode())
|
||||
.setAlarmTypeName(nullToEmpty(p.alarmTypeName()));
|
||||
if (p.faultCodes() != null) b.addAllFaultCodes(p.faultCodes());
|
||||
if (p.activeBits() != null) b.addAllActiveBits(p.activeBits());
|
||||
if (p.longitude() != null) b.setLongitude(p.longitude());
|
||||
if (p.latitude() != null) b.setLatitude(p.latitude());
|
||||
return b.build();
|
||||
}
|
||||
|
||||
private static String nullToEmpty(String s) {
|
||||
return s == null ? "" : s;
|
||||
}
|
||||
|
||||
private static String rawArchiveKey(VehicleEvent.RawArchive raw) {
|
||||
String key = raw.metadata() == null ? "" : raw.metadata().getOrDefault(RawArchiveKeys.META_KEY, "");
|
||||
return key == null || key.isBlank() ? RawArchiveKeys.key(raw) : key;
|
||||
}
|
||||
|
||||
private static String rawArchiveUri(VehicleEvent.RawArchive raw, String key) {
|
||||
String uri = raw.metadata() == null ? "" : raw.metadata().getOrDefault(RawArchiveKeys.META_URI, "");
|
||||
return uri == null || uri.isBlank() ? RawArchiveKeys.logicalUri(key) : uri;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
package com.lingniu.ingest.sink.mq;
|
||||
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
|
||||
import org.apache.kafka.clients.consumer.ConsumerConfig;
|
||||
import org.apache.kafka.clients.consumer.KafkaConsumer;
|
||||
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
|
||||
import org.apache.kafka.common.serialization.StringDeserializer;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Properties;
|
||||
import java.util.function.Function;
|
||||
|
||||
public final class KafkaEnvelopeConsumerFactory {
|
||||
|
||||
private final Function<Properties, org.apache.kafka.clients.consumer.Consumer<String, byte[]>> consumerFactory;
|
||||
|
||||
public KafkaEnvelopeConsumerFactory() {
|
||||
this(KafkaConsumer::new);
|
||||
}
|
||||
|
||||
KafkaEnvelopeConsumerFactory(
|
||||
Function<Properties, org.apache.kafka.clients.consumer.Consumer<String, byte[]>> consumerFactory) {
|
||||
this.consumerFactory = consumerFactory;
|
||||
}
|
||||
|
||||
public List<KafkaEnvelopeConsumerWorker> createWorkers(Map<String, EnvelopeConsumerProcessor> processors,
|
||||
SinkMqProperties props) {
|
||||
Map<String, SinkMqProperties.Binding> bindings = effectiveBindings(props);
|
||||
List<KafkaEnvelopeConsumerWorker> workers = new ArrayList<>();
|
||||
for (Map.Entry<String, SinkMqProperties.Binding> entry : bindings.entrySet()) {
|
||||
String processorBeanName = entry.getKey();
|
||||
EnvelopeConsumerProcessor processor = processors.get(processorBeanName);
|
||||
SinkMqProperties.Binding binding = entry.getValue();
|
||||
if (processor == null || binding == null || !binding.isEnabled()) {
|
||||
continue;
|
||||
}
|
||||
List<String> topics = cleanTopics(binding.getTopics());
|
||||
if (topics.isEmpty()) {
|
||||
continue;
|
||||
}
|
||||
KafkaEnvelopeConsumerWorker worker = new KafkaEnvelopeConsumerWorker(
|
||||
consumerFactory.apply(consumerProperties(props, binding, processorBeanName)),
|
||||
topicProcessors(topics, processor));
|
||||
worker.subscribe(topics);
|
||||
workers.add(worker);
|
||||
}
|
||||
return workers;
|
||||
}
|
||||
|
||||
private Map<String, SinkMqProperties.Binding> effectiveBindings(SinkMqProperties props) {
|
||||
Map<String, SinkMqProperties.Binding> configured = props.getConsumer().getBindings();
|
||||
if (configured != null && !configured.isEmpty()) {
|
||||
return configured;
|
||||
}
|
||||
SinkMqProperties.Topics topics = props.getTopics();
|
||||
Map<String, SinkMqProperties.Binding> defaults = new LinkedHashMap<>();
|
||||
defaults.put("eventHistoryEnvelopeConsumerProcessor", binding(
|
||||
"vehicle-event-history",
|
||||
topics.getRealtime(), topics.getLocation(), topics.getAlarm(), topics.getSession(), topics.getMediaMeta()));
|
||||
defaults.put("vehicleStateEnvelopeConsumerProcessor", binding(
|
||||
"vehicle-state",
|
||||
topics.getRealtime(), topics.getLocation(), topics.getAlarm()));
|
||||
defaults.put("vehicleStatEnvelopeConsumerProcessor", binding(
|
||||
"vehicle-stat",
|
||||
topics.getRealtime(), topics.getLocation()));
|
||||
return defaults;
|
||||
}
|
||||
|
||||
private SinkMqProperties.Binding binding(String groupId, String... topics) {
|
||||
SinkMqProperties.Binding binding = new SinkMqProperties.Binding();
|
||||
binding.setGroupId(groupId);
|
||||
binding.setTopics(List.of(topics));
|
||||
return binding;
|
||||
}
|
||||
|
||||
private Map<String, EnvelopeConsumerProcessor> topicProcessors(List<String> topics, EnvelopeConsumerProcessor processor) {
|
||||
Map<String, EnvelopeConsumerProcessor> byTopic = new LinkedHashMap<>();
|
||||
for (String topic : topics) {
|
||||
byTopic.put(topic, processor);
|
||||
}
|
||||
return byTopic;
|
||||
}
|
||||
|
||||
private List<String> cleanTopics(List<String> topics) {
|
||||
if (topics == null || topics.isEmpty()) {
|
||||
return List.of();
|
||||
}
|
||||
LinkedHashSet<String> clean = new LinkedHashSet<>();
|
||||
for (String topic : topics) {
|
||||
if (topic != null && !topic.isBlank()) {
|
||||
clean.add(topic);
|
||||
}
|
||||
}
|
||||
return List.copyOf(clean);
|
||||
}
|
||||
|
||||
private Properties consumerProperties(SinkMqProperties props,
|
||||
SinkMqProperties.Binding binding,
|
||||
String processorBeanName) {
|
||||
SinkMqProperties.Consumer consumer = props.getConsumer();
|
||||
Properties p = new Properties();
|
||||
p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, props.getBootstrapServers());
|
||||
p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
|
||||
p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
|
||||
p.put(ConsumerConfig.GROUP_ID_CONFIG, groupId(binding, processorBeanName));
|
||||
p.put(ConsumerConfig.CLIENT_ID_CONFIG, consumer.getClientIdPrefix() + "-" + processorBeanName);
|
||||
p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
|
||||
p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, consumer.getAutoOffsetReset());
|
||||
p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, consumer.getMaxPollRecords());
|
||||
return p;
|
||||
}
|
||||
|
||||
private String groupId(SinkMqProperties.Binding binding, String processorBeanName) {
|
||||
if (binding.getGroupId() != null && !binding.getGroupId().isBlank()) {
|
||||
return binding.getGroupId();
|
||||
}
|
||||
return processorBeanName.replace("EnvelopeConsumerProcessor", "");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,125 @@
|
||||
package com.lingniu.ingest.sink.mq;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.context.SmartLifecycle;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
public final class KafkaEnvelopeConsumerRunner implements SmartLifecycle, AutoCloseable {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(KafkaEnvelopeConsumerRunner.class);
|
||||
|
||||
private final List<KafkaEnvelopeConsumerWorker> workers;
|
||||
private final Duration pollTimeout;
|
||||
private final Duration loopBackoff;
|
||||
private final boolean autoStartup;
|
||||
private final AtomicBoolean running = new AtomicBoolean(false);
|
||||
private final AtomicBoolean closed = new AtomicBoolean(false);
|
||||
private ExecutorService executor;
|
||||
|
||||
public KafkaEnvelopeConsumerRunner(List<KafkaEnvelopeConsumerWorker> workers,
|
||||
Duration pollTimeout,
|
||||
Duration loopBackoff,
|
||||
boolean autoStartup) {
|
||||
if (workers == null || workers.isEmpty()) {
|
||||
throw new IllegalArgumentException("workers must not be empty");
|
||||
}
|
||||
this.workers = List.copyOf(workers);
|
||||
this.pollTimeout = pollTimeout == null ? Duration.ofSeconds(1) : pollTimeout;
|
||||
this.loopBackoff = loopBackoff == null ? Duration.ofSeconds(1) : loopBackoff;
|
||||
this.autoStartup = autoStartup;
|
||||
}
|
||||
|
||||
public List<KafkaEnvelopeConsumerWorker> workers() {
|
||||
return workers;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start() {
|
||||
if (!running.compareAndSet(false, true)) {
|
||||
return;
|
||||
}
|
||||
executor = Executors.newFixedThreadPool(workers.size(), r -> {
|
||||
Thread thread = new Thread(r, "kafka-envelope-consumer");
|
||||
thread.setDaemon(true);
|
||||
return thread;
|
||||
});
|
||||
for (KafkaEnvelopeConsumerWorker worker : workers) {
|
||||
executor.submit(() -> pollLoop(worker));
|
||||
}
|
||||
}
|
||||
|
||||
private void pollLoop(KafkaEnvelopeConsumerWorker worker) {
|
||||
while (running.get()) {
|
||||
try {
|
||||
worker.pollOnce(pollTimeout);
|
||||
} catch (RuntimeException ex) {
|
||||
log.warn("Kafka envelope consumer poll failed; the worker will retry after backoff", ex);
|
||||
sleepBackoff();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void sleepBackoff() {
|
||||
try {
|
||||
Thread.sleep(loopBackoff.toMillis());
|
||||
} catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
if (!running.compareAndSet(true, false)) {
|
||||
return;
|
||||
}
|
||||
if (executor != null) {
|
||||
executor.shutdownNow();
|
||||
}
|
||||
closeWorkers();
|
||||
if (executor != null) {
|
||||
try {
|
||||
executor.awaitTermination(5, TimeUnit.SECONDS);
|
||||
} catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop(Runnable callback) {
|
||||
stop();
|
||||
callback.run();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return running.get();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isAutoStartup() {
|
||||
return autoStartup;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
stop();
|
||||
closeWorkers();
|
||||
}
|
||||
|
||||
private void closeWorkers() {
|
||||
if (!closed.compareAndSet(false, true)) {
|
||||
return;
|
||||
}
|
||||
for (KafkaEnvelopeConsumerWorker worker : workers) {
|
||||
worker.close();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
package com.lingniu.ingest.sink.mq;
|
||||
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeConsumerRecord;
|
||||
import org.apache.kafka.clients.consumer.Consumer;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecords;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Collection;
|
||||
import java.util.Map;
|
||||
|
||||
public final class KafkaEnvelopeConsumerWorker implements AutoCloseable {
|
||||
|
||||
private final Consumer<String, byte[]> consumer;
|
||||
private final Map<String, EnvelopeConsumerProcessor> processorsByTopic;
|
||||
|
||||
public KafkaEnvelopeConsumerWorker(Consumer<String, byte[]> consumer,
|
||||
Map<String, EnvelopeConsumerProcessor> processorsByTopic) {
|
||||
if (consumer == null) {
|
||||
throw new IllegalArgumentException("consumer must not be null");
|
||||
}
|
||||
if (processorsByTopic == null || processorsByTopic.isEmpty()) {
|
||||
throw new IllegalArgumentException("processorsByTopic must not be empty");
|
||||
}
|
||||
this.consumer = consumer;
|
||||
this.processorsByTopic = Map.copyOf(processorsByTopic);
|
||||
}
|
||||
|
||||
public void subscribe(Collection<String> topics) {
|
||||
consumer.subscribe(topics);
|
||||
}
|
||||
|
||||
public int pollOnce(Duration timeout) {
|
||||
ConsumerRecords<String, byte[]> records = consumer.poll(timeout == null ? Duration.ZERO : timeout);
|
||||
int processed = 0;
|
||||
for (ConsumerRecord<String, byte[]> record : records) {
|
||||
EnvelopeConsumerProcessor processor = processorsByTopic.get(record.topic());
|
||||
if (processor == null) {
|
||||
continue;
|
||||
}
|
||||
processor.process(new EnvelopeConsumerRecord(
|
||||
record.topic(),
|
||||
record.partition(),
|
||||
record.offset(),
|
||||
record.key(),
|
||||
record.value()));
|
||||
processed++;
|
||||
}
|
||||
if (processed > 0) {
|
||||
consumer.commitSync();
|
||||
}
|
||||
return processed;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
consumer.close();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
package com.lingniu.ingest.sink.mq;
|
||||
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterRecord;
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterSink;
|
||||
import org.apache.kafka.clients.producer.KafkaProducer;
|
||||
import org.apache.kafka.clients.producer.Producer;
|
||||
import org.apache.kafka.clients.producer.ProducerRecord;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
public final class KafkaEnvelopeDeadLetterSink implements EnvelopeDeadLetterSink {
|
||||
|
||||
private final Producer<String, byte[]> producer;
|
||||
private final String topic;
|
||||
|
||||
public KafkaEnvelopeDeadLetterSink(KafkaProducer<String, byte[]> producer, String topic) {
|
||||
this((Producer<String, byte[]>) producer, topic);
|
||||
}
|
||||
|
||||
KafkaEnvelopeDeadLetterSink(Producer<String, byte[]> producer, String topic) {
|
||||
if (producer == null) {
|
||||
throw new IllegalArgumentException("producer must not be null");
|
||||
}
|
||||
if (topic == null || topic.isBlank()) {
|
||||
throw new IllegalArgumentException("topic must not be blank");
|
||||
}
|
||||
this.producer = producer;
|
||||
this.topic = topic;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void publish(EnvelopeDeadLetterRecord record) {
|
||||
if (record == null) {
|
||||
throw new IllegalArgumentException("record must not be null");
|
||||
}
|
||||
ProducerRecord<String, byte[]> out = new ProducerRecord<>(topic, record.key(), record.payload());
|
||||
header(out, "dlq-service", record.service());
|
||||
header(out, "dlq-source-topic", record.topic());
|
||||
header(out, "dlq-source-partition", Integer.toString(record.partition()));
|
||||
header(out, "dlq-source-offset", Long.toString(record.offset()));
|
||||
header(out, "dlq-status", record.status().name());
|
||||
header(out, "dlq-event-id", record.eventId());
|
||||
header(out, "dlq-vin", record.vin());
|
||||
header(out, "dlq-message", record.message());
|
||||
header(out, "dlq-created-at", record.createdAt().toString());
|
||||
producer.send(out);
|
||||
}
|
||||
|
||||
private static void header(ProducerRecord<String, byte[]> record, String key, String value) {
|
||||
record.headers().add(key, value.getBytes(StandardCharsets.UTF_8));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
package com.lingniu.ingest.sink.mq;
|
||||
|
||||
import com.lingniu.ingest.api.event.VehicleEvent;
|
||||
import com.lingniu.ingest.api.sink.EventSink;
|
||||
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
|
||||
import org.apache.kafka.clients.producer.KafkaProducer;
|
||||
import org.apache.kafka.clients.producer.ProducerRecord;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
/**
|
||||
* Kafka Sink：序列化为 Protobuf Envelope，按 vin 分区，异步发送。
|
||||
*
|
||||
* <p>失败处理：Resilience4j 熔断；熔断开启时事件转投 DLQ topic。
|
||||
*/
|
||||
public final class KafkaEventSink implements EventSink, AutoCloseable {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(KafkaEventSink.class);
|
||||
|
||||
private final KafkaProducer<String, byte[]> producer;
|
||||
private final EnvelopeMapper mapper;
|
||||
private final TopicRouter router;
|
||||
private final String dlqTopic;
|
||||
private final CircuitBreaker breaker;
|
||||
|
||||
public KafkaEventSink(KafkaProducer<String, byte[]> producer,
|
||||
EnvelopeMapper mapper,
|
||||
TopicRouter router,
|
||||
String dlqTopic,
|
||||
CircuitBreaker breaker) {
|
||||
this.producer = producer;
|
||||
this.mapper = mapper;
|
||||
this.router = router;
|
||||
this.dlqTopic = dlqTopic;
|
||||
this.breaker = breaker;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String name() {
|
||||
return "kafka";
|
||||
}
|
||||
|
||||
/**
|
||||
* 拒绝 {@link VehicleEvent.RawArchive} —— 原始报文体积大、属于冷存域，本期由
|
||||
* {@code ArchiveEventSink} 独占处理。未来若需要将 raw archive URI 回填到 Kafka 的
|
||||
* {@code vehicle.raw.archive} topic，再放开此过滤并在 Envelope 里填 uri/checksum。
|
||||
*/
|
||||
@Override
|
||||
public boolean accepts(VehicleEvent event) {
|
||||
return !(event instanceof VehicleEvent.RawArchive);
|
||||
}
|
||||
|
||||
@Override
|
||||
public CompletableFuture<Void> publish(VehicleEvent event) {
|
||||
CompletableFuture<Void> cf = new CompletableFuture<>();
|
||||
byte[] payload;
|
||||
try {
|
||||
payload = mapper.toEnvelope(event).toByteArray();
|
||||
} catch (Exception e) {
|
||||
log.error("envelope build failed eventId={} vin={}", event.eventId(), event.vin(), e);
|
||||
cf.completeExceptionally(e);
|
||||
return cf;
|
||||
}
|
||||
|
||||
String topic = breaker.tryAcquirePermission() ? router.route(event) : dlqTopic;
|
||||
ProducerRecord<String, byte[]> record = new ProducerRecord<>(topic, event.vin(), payload);
|
||||
record.headers().add("event-id", event.eventId().getBytes());
|
||||
record.headers().add("trace-id", event.traceId() == null ? new byte[0] : event.traceId().getBytes());
|
||||
record.headers().add("source", event.source().name().getBytes());
|
||||
|
||||
producer.send(record, (metadata, ex) -> {
|
||||
if (ex != null) {
|
||||
breaker.onError(0, java.util.concurrent.TimeUnit.MILLISECONDS, ex);
|
||||
cf.completeExceptionally(ex);
|
||||
} else {
|
||||
breaker.onSuccess(0, java.util.concurrent.TimeUnit.MILLISECONDS);
|
||||
cf.complete(null);
|
||||
}
|
||||
});
|
||||
return cf;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
producer.flush();
|
||||
producer.close();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
package com.lingniu.ingest.sink.mq;
|
||||
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
|
||||
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
|
||||
import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
|
||||
import org.apache.kafka.clients.producer.KafkaProducer;
|
||||
import org.apache.kafka.clients.producer.ProducerConfig;
|
||||
import org.apache.kafka.common.serialization.ByteArraySerializer;
|
||||
import org.apache.kafka.common.serialization.StringSerializer;
|
||||
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Properties;
|
||||
|
||||
/**
|
||||
* MQ Sink 自动装配。
|
||||
*
|
||||
* <p>装配前提（两个条件都要满足）：
|
||||
* <ol>
|
||||
* <li>{@code lingniu.ingest.sink.mq.enabled=true}（默认 true）—— <b>总开关</b>，
|
||||
* 设为 false 时本模块完全不装配，ingest-core 的 DisruptorEventBus 仍然运行但
|
||||
* 没有 Kafka sink。
|
||||
* <li>{@code lingniu.ingest.sink.mq.type=kafka}（默认 kafka）—— 后端选择，预留
|
||||
* rocketmq/pulsar 等扩展。
|
||||
* </ol>
|
||||
*/
|
||||
@AutoConfiguration
|
||||
@EnableConfigurationProperties(SinkMqProperties.class)
|
||||
@ConditionalOnProperty(prefix = "lingniu.ingest.sink.mq", name = "enabled", havingValue = "true", matchIfMissing = true)
|
||||
public class SinkMqAutoConfiguration {
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public EnvelopeMapper envelopeMapper(SinkMqProperties props) {
|
||||
return new EnvelopeMapper(props.getNodeId());
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public TopicRouter topicRouter(SinkMqProperties props) {
|
||||
return new TopicRouter(props.getTopics());
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
@ConditionalOnProperty(prefix = "lingniu.ingest.sink.mq", name = "type", havingValue = "kafka", matchIfMissing = true)
|
||||
public CircuitBreaker kafkaSinkCircuitBreaker() {
|
||||
return CircuitBreaker.of("kafka-sink", CircuitBreakerConfig.custom()
|
||||
.slidingWindowSize(100)
|
||||
.failureRateThreshold(50)
|
||||
.waitDurationInOpenState(Duration.ofSeconds(10))
|
||||
.permittedNumberOfCallsInHalfOpenState(10)
|
||||
.build());
|
||||
}
|
||||
|
||||
@Bean(destroyMethod = "close")
|
||||
@ConditionalOnMissingBean
|
||||
@ConditionalOnProperty(prefix = "lingniu.ingest.sink.mq", name = "type", havingValue = "kafka", matchIfMissing = true)
|
||||
public KafkaProducer<String, byte[]> kafkaProducer(SinkMqProperties props) {
|
||||
Properties p = new Properties();
|
||||
p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, props.getBootstrapServers());
|
||||
p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
|
||||
p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
|
||||
p.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, props.getCompressionType());
|
||||
p.put(ProducerConfig.LINGER_MS_CONFIG, props.getLingerMs());
|
||||
p.put(ProducerConfig.BATCH_SIZE_CONFIG, props.getBatchSize());
|
||||
p.put(ProducerConfig.ACKS_CONFIG, props.getAcks());
|
||||
p.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, props.isEnableIdempotence());
|
||||
return new KafkaProducer<>(p);
|
||||
}
|
||||
|
||||
@Bean(destroyMethod = "close")
|
||||
@ConditionalOnMissingBean
|
||||
@ConditionalOnProperty(prefix = "lingniu.ingest.sink.mq", name = "type", havingValue = "kafka", matchIfMissing = true)
|
||||
public KafkaEventSink kafkaEventSink(KafkaProducer<String, byte[]> producer,
|
||||
EnvelopeMapper mapper,
|
||||
TopicRouter router,
|
||||
SinkMqProperties props,
|
||||
CircuitBreaker breaker) {
|
||||
return new KafkaEventSink(producer, mapper, router, props.getTopics().getDlq(), breaker);
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
@ConditionalOnProperty(prefix = "lingniu.ingest.sink.mq", name = "type", havingValue = "kafka", matchIfMissing = true)
|
||||
public KafkaEnvelopeDeadLetterSink kafkaEnvelopeDeadLetterSink(KafkaProducer<String, byte[]> producer,
|
||||
SinkMqProperties props) {
|
||||
return new KafkaEnvelopeDeadLetterSink(producer, props.getTopics().getDlq());
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
@ConditionalOnBean(EnvelopeConsumerProcessor.class)
|
||||
@ConditionalOnProperty(prefix = "lingniu.ingest.sink.mq.consumer", name = "enabled", havingValue = "true")
|
||||
public KafkaEnvelopeConsumerRunner kafkaEnvelopeConsumerRunner(Map<String, EnvelopeConsumerProcessor> processors,
|
||||
SinkMqProperties props) {
|
||||
List<KafkaEnvelopeConsumerWorker> workers = new KafkaEnvelopeConsumerFactory().createWorkers(processors, props);
|
||||
if (workers.isEmpty()) {
|
||||
throw new IllegalStateException("no kafka envelope consumer workers created; check consumer bindings");
|
||||
}
|
||||
return new KafkaEnvelopeConsumerRunner(
|
||||
workers,
|
||||
Duration.ofMillis(props.getConsumer().getPollTimeoutMillis()),
|
||||
Duration.ofMillis(props.getConsumer().getLoopBackoffMillis()),
|
||||
props.getConsumer().isAutoStartup());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
package com.lingniu.ingest.sink.mq;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
@ConfigurationProperties(prefix = "lingniu.ingest.sink.mq")
|
||||
public class SinkMqProperties {
|
||||
|
||||
/**
|
||||
* MQ Sink 总开关。默认 {@code true}。
|
||||
* 设为 {@code false} 时 {@link SinkMqAutoConfiguration} 完全不装配任何 Bean（Kafka Producer、
|
||||
* EnvelopeMapper、TopicRouter、KafkaEventSink 都不会创建），ingest-core 的 DisruptorEventBus
|
||||
* 仍然正常运行但没有外部 sink（事件落地到 sink-archive 或 Noop 吞掉）。
|
||||
*/
|
||||
private boolean enabled = true;
|
||||
|
||||
/** MQ 后端类型。目前仅支持 kafka，预留 rocketmq/pulsar 等。 */
|
||||
private String type = "kafka";
|
||||
private String bootstrapServers = "114.55.58.251:9092";
|
||||
private String compressionType = "zstd";
|
||||
private int lingerMs = 20;
|
||||
private int batchSize = 65536;
|
||||
private String acks = "all";
|
||||
private boolean enableIdempotence = true;
|
||||
private String nodeId = "ingest-local";
|
||||
private Topics topics = new Topics();
|
||||
private Consumer consumer = new Consumer();
|
||||
|
||||
public boolean isEnabled() { return enabled; }
|
||||
public void setEnabled(boolean enabled) { this.enabled = enabled; }
|
||||
public String getType() { return type; }
|
||||
public void setType(String type) { this.type = type; }
|
||||
public String getBootstrapServers() { return bootstrapServers; }
|
||||
public void setBootstrapServers(String bootstrapServers) { this.bootstrapServers = bootstrapServers; }
|
||||
public String getCompressionType() { return compressionType; }
|
||||
public void setCompressionType(String compressionType) { this.compressionType = compressionType; }
|
||||
public int getLingerMs() { return lingerMs; }
|
||||
public void setLingerMs(int lingerMs) { this.lingerMs = lingerMs; }
|
||||
public int getBatchSize() { return batchSize; }
|
||||
public void setBatchSize(int batchSize) { this.batchSize = batchSize; }
|
||||
public String getAcks() { return acks; }
|
||||
public void setAcks(String acks) { this.acks = acks; }
|
||||
public boolean isEnableIdempotence() { return enableIdempotence; }
|
||||
public void setEnableIdempotence(boolean enableIdempotence) { this.enableIdempotence = enableIdempotence; }
|
||||
public String getNodeId() { return nodeId; }
|
||||
public void setNodeId(String nodeId) { this.nodeId = nodeId; }
|
||||
public Topics getTopics() { return topics; }
|
||||
public void setTopics(Topics topics) { this.topics = topics; }
|
||||
public Consumer getConsumer() { return consumer; }
|
||||
public void setConsumer(Consumer consumer) { this.consumer = consumer; }
|
||||
|
||||
public static class Topics {
|
||||
private String realtime = "vehicle.realtime";
|
||||
private String location = "vehicle.location";
|
||||
private String alarm = "vehicle.alarm";
|
||||
private String session = "vehicle.session";
|
||||
private String mediaMeta = "vehicle.media.meta";
|
||||
private String rawArchive = "vehicle.raw.archive";
|
||||
private String dlq = "vehicle.dlq";
|
||||
|
||||
public String getRealtime() { return realtime; }
|
||||
public void setRealtime(String realtime) { this.realtime = realtime; }
|
||||
public String getLocation() { return location; }
|
||||
public void setLocation(String location) { this.location = location; }
|
||||
public String getAlarm() { return alarm; }
|
||||
public void setAlarm(String alarm) { this.alarm = alarm; }
|
||||
public String getSession() { return session; }
|
||||
public void setSession(String session) { this.session = session; }
|
||||
public String getMediaMeta() { return mediaMeta; }
|
||||
public void setMediaMeta(String mediaMeta) { this.mediaMeta = mediaMeta; }
|
||||
public String getRawArchive() { return rawArchive; }
|
||||
public void setRawArchive(String rawArchive) { this.rawArchive = rawArchive; }
|
||||
public String getDlq() { return dlq; }
|
||||
public void setDlq(String dlq) { this.dlq = dlq; }
|
||||
}
|
||||
|
||||
public static class Consumer {
|
||||
private boolean enabled = false;
|
||||
private boolean autoStartup = true;
|
||||
private String clientIdPrefix = "lingniu-envelope-consumer";
|
||||
private int pollTimeoutMillis = 1000;
|
||||
private int loopBackoffMillis = 1000;
|
||||
private String autoOffsetReset = "earliest";
|
||||
private int maxPollRecords = 500;
|
||||
private Map<String, Binding> bindings = new LinkedHashMap<>();
|
||||
|
||||
public boolean isEnabled() { return enabled; }
|
||||
public void setEnabled(boolean enabled) { this.enabled = enabled; }
|
||||
public boolean isAutoStartup() { return autoStartup; }
|
||||
public void setAutoStartup(boolean autoStartup) { this.autoStartup = autoStartup; }
|
||||
public String getClientIdPrefix() { return clientIdPrefix; }
|
||||
public void setClientIdPrefix(String clientIdPrefix) { this.clientIdPrefix = clientIdPrefix; }
|
||||
public int getPollTimeoutMillis() { return pollTimeoutMillis; }
|
||||
public void setPollTimeoutMillis(int pollTimeoutMillis) { this.pollTimeoutMillis = pollTimeoutMillis; }
|
||||
public int getLoopBackoffMillis() { return loopBackoffMillis; }
|
||||
public void setLoopBackoffMillis(int loopBackoffMillis) { this.loopBackoffMillis = loopBackoffMillis; }
|
||||
public String getAutoOffsetReset() { return autoOffsetReset; }
|
||||
public void setAutoOffsetReset(String autoOffsetReset) { this.autoOffsetReset = autoOffsetReset; }
|
||||
public int getMaxPollRecords() { return maxPollRecords; }
|
||||
public void setMaxPollRecords(int maxPollRecords) { this.maxPollRecords = maxPollRecords; }
|
||||
public Map<String, Binding> getBindings() { return bindings; }
|
||||
public void setBindings(Map<String, Binding> bindings) { this.bindings = bindings; }
|
||||
}
|
||||
|
||||
public static class Binding {
|
||||
private boolean enabled = true;
|
||||
private String groupId;
|
||||
private List<String> topics = new ArrayList<>();
|
||||
|
||||
public boolean isEnabled() { return enabled; }
|
||||
public void setEnabled(boolean enabled) { this.enabled = enabled; }
|
||||
public String getGroupId() { return groupId; }
|
||||
public void setGroupId(String groupId) { this.groupId = groupId; }
|
||||
public List<String> getTopics() { return topics; }
|
||||
public void setTopics(List<String> topics) { this.topics = topics; }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
package com.lingniu.ingest.sink.mq;
|
||||
|
||||
import com.lingniu.ingest.api.event.VehicleEvent;
|
||||
|
||||
/**
|
||||
* 事件 → Kafka topic 路由。集中管理避免散落。
|
||||
*/
|
||||
public final class TopicRouter {
|
||||
|
||||
private final SinkMqProperties.Topics topics;
|
||||
|
||||
public TopicRouter(SinkMqProperties.Topics topics) {
|
||||
this.topics = topics;
|
||||
}
|
||||
|
||||
public String route(VehicleEvent event) {
|
||||
return switch (event) {
|
||||
case VehicleEvent.Realtime _ -> topics.getRealtime();
|
||||
case VehicleEvent.Location _ -> topics.getLocation();
|
||||
case VehicleEvent.Alarm _ -> topics.getAlarm();
|
||||
case VehicleEvent.Login _,
|
||||
VehicleEvent.Logout _,
|
||||
VehicleEvent.Heartbeat _ -> topics.getSession();
|
||||
case VehicleEvent.MediaMeta _ -> topics.getMediaMeta();
|
||||
case VehicleEvent.Passthrough _ -> topics.getAlarm();
|
||||
case VehicleEvent.RawArchive _ -> topics.getRawArchive();
|
||||
};
|
||||
}
|
||||
}
|
||||
128
modules/sinks/sink-mq/src/main/proto/vehicle_envelope.proto
Normal file
128
modules/sinks/sink-mq/src/main/proto/vehicle_envelope.proto
Normal file
@@ -0,0 +1,128 @@
|
||||
syntax = "proto3";
|
||||
|
||||
package com.lingniu.ingest.sink.mq.proto;
|
||||
|
||||
option java_multiple_files = true;
|
||||
option java_package = "com.lingniu.ingest.sink.mq.proto";
|
||||
option java_outer_classname = "VehicleEnvelopeProto";
|
||||
|
||||
// 统一消息外壳：所有 Topic 共用此 Envelope，payload 通过 oneof 区分具体事件类型。
|
||||
message VehicleEnvelope {
|
||||
string schema_version = 1;
|
||||
string event_id = 2;
|
||||
string trace_id = 3;
|
||||
string vin = 4;
|
||||
string source = 5; // ProtocolId 字符串形式
|
||||
string protocol_version = 6;
|
||||
int64 event_time_ms = 7;
|
||||
int64 ingest_time_ms = 8;
|
||||
string ingest_node_id = 9;
|
||||
map<string, string> metadata = 10;
|
||||
|
||||
oneof payload {
|
||||
RealtimePayload realtime = 20;
|
||||
LocationPayload location = 21;
|
||||
AlarmPayload alarm = 22;
|
||||
LoginPayload login = 23;
|
||||
LogoutPayload logout = 24;
|
||||
HeartbeatPayload heartbeat = 25;
|
||||
MediaMetaPayload media_meta = 26;
|
||||
PassthroughPayload passthrough = 27;
|
||||
}
|
||||
|
||||
// 可选：原始字节指针
|
||||
RawArchiveRef raw_archive = 40;
|
||||
|
||||
// 全字段内部遥测快照。新下游消费者优先读取这里，避免依赖协议字段名。
|
||||
TelemetrySnapshot telemetry_snapshot = 50;
|
||||
}
|
||||
|
||||
message RawArchiveRef {
|
||||
string uri = 1;
|
||||
string checksum = 2;
|
||||
int64 size_bytes = 3;
|
||||
}
|
||||
|
||||
message TelemetrySnapshot {
|
||||
string event_type = 1;
|
||||
string raw_archive_uri = 2;
|
||||
repeated TelemetryField fields = 3;
|
||||
}
|
||||
|
||||
message TelemetryField {
|
||||
string key = 1;
|
||||
string value_type = 2;
|
||||
string value = 3;
|
||||
string unit = 4;
|
||||
string quality = 5;
|
||||
string source_path = 6;
|
||||
}
|
||||
|
||||
message RealtimePayload {
|
||||
optional double speed_kmh = 1;
|
||||
optional double total_mileage_km = 2;
|
||||
optional double battery_soc = 3;
|
||||
optional double battery_voltage_v = 4;
|
||||
optional double battery_current_a = 5;
|
||||
optional double fc_voltage_v = 6;
|
||||
optional double fc_current_a = 7;
|
||||
optional double fc_temp_c = 8;
|
||||
optional double hydrogen_remaining_kg = 9;
|
||||
optional double hydrogen_high_pressure_mpa = 10;
|
||||
optional double hydrogen_low_pressure_mpa = 11;
|
||||
optional string vehicle_state = 12;
|
||||
optional string charging_state = 13;
|
||||
optional string running_mode = 14;
|
||||
optional int32 gear_level = 15;
|
||||
optional double accelerator_pedal = 16;
|
||||
optional double brake_pedal = 17;
|
||||
optional double longitude = 18;
|
||||
optional double latitude = 19;
|
||||
optional double altitude_m = 20;
|
||||
optional double direction_deg = 21;
|
||||
optional double ambient_temp_c = 22;
|
||||
optional double coolant_temp_c = 23;
|
||||
}
|
||||
|
||||
message LocationPayload {
|
||||
double longitude = 1;
|
||||
double latitude = 2;
|
||||
double altitude_m = 3;
|
||||
double speed_kmh = 4;
|
||||
double direction_deg = 5;
|
||||
int64 alarm_flag = 6;
|
||||
int64 status_flag = 7;
|
||||
}
|
||||
|
||||
message AlarmPayload {
|
||||
string level = 1;
|
||||
int32 alarm_type_code = 2;
|
||||
string alarm_type_name = 3;
|
||||
repeated string fault_codes = 4;
|
||||
optional double longitude = 5;
|
||||
optional double latitude = 6;
|
||||
// 通用报警标志位（2016 版 0~15 / 2025 版 0~27），对照 GB/T 32960.3 表 24。
|
||||
// 例：SOC_LOW / BATTERY_HIGH_TEMP / HYDROGEN_LEAK
|
||||
repeated string active_bits = 7;
|
||||
}
|
||||
|
||||
message LoginPayload {
|
||||
string iccid = 1;
|
||||
string protocol_version = 2;
|
||||
}
|
||||
|
||||
message LogoutPayload {}
|
||||
|
||||
message HeartbeatPayload {}
|
||||
|
||||
message MediaMetaPayload {
|
||||
string media_id = 1;
|
||||
string media_type = 2;
|
||||
int64 size_bytes = 3;
|
||||
string archive_ref = 4;
|
||||
}
|
||||
|
||||
message PassthroughPayload {
|
||||
int32 passthrough_type = 1;
|
||||
bytes data = 2;
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
com.lingniu.ingest.sink.mq.SinkMqAutoConfiguration
|
||||
@@ -0,0 +1,94 @@
|
||||
package com.lingniu.ingest.sink.mq;
|
||||
|
||||
import com.lingniu.ingest.api.ProtocolId;
|
||||
import com.lingniu.ingest.api.event.RawArchiveKeys;
|
||||
import com.lingniu.ingest.api.event.RealtimePayload;
|
||||
import com.lingniu.ingest.api.event.VehicleEvent;
|
||||
import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
class EnvelopeMapperTelemetrySnapshotTest {
|
||||
|
||||
@Test
|
||||
void mapsVehicleEventToFullFieldTelemetrySnapshot() {
|
||||
VehicleEvent.Realtime event = new VehicleEvent.Realtime(
|
||||
"event-1",
|
||||
"VIN001",
|
||||
ProtocolId.GB32960,
|
||||
Instant.parse("2026-06-22T08:00:00Z"),
|
||||
Instant.parse("2026-06-22T08:00:01Z"),
|
||||
"trace-1",
|
||||
Map.of("protocolVersion", "V2016", "rawArchiveUri", "archive://raw/event-1.bin"),
|
||||
new RealtimePayload(
|
||||
80.5,
|
||||
12345.6,
|
||||
55.0,
|
||||
610.0,
|
||||
-20.5,
|
||||
null,
|
||||
null,
|
||||
null,
|
||||
8.2,
|
||||
35.0,
|
||||
null,
|
||||
RealtimePayload.VehicleState.STARTED,
|
||||
RealtimePayload.ChargingState.UNCHARGED,
|
||||
RealtimePayload.RunningMode.FUEL,
|
||||
3,
|
||||
null,
|
||||
null,
|
||||
113.12,
|
||||
23.45,
|
||||
null,
|
||||
null,
|
||||
null,
|
||||
null
|
||||
)
|
||||
);
|
||||
|
||||
VehicleEnvelope envelope = new EnvelopeMapper("node-1").toEnvelope(event);
|
||||
|
||||
assertThat(envelope.getTelemetrySnapshot().getEventType()).isEqualTo("REALTIME");
|
||||
assertThat(envelope.getTelemetrySnapshot().getRawArchiveUri()).isEqualTo("archive://raw/event-1.bin");
|
||||
assertThat(envelope.getTelemetrySnapshot().getFieldsList())
|
||||
.anySatisfy(field -> {
|
||||
assertThat(field.getKey()).isEqualTo("total_mileage_km");
|
||||
assertThat(field.getValueType()).isEqualTo("DOUBLE");
|
||||
assertThat(field.getValue()).isEqualTo("12345.6");
|
||||
assertThat(field.getUnit()).isEqualTo("km");
|
||||
})
|
||||
.anySatisfy(field -> {
|
||||
assertThat(field.getKey()).isEqualTo("hydrogen_high_pressure_mpa");
|
||||
assertThat(field.getValue()).isEqualTo("35.0");
|
||||
assertThat(field.getUnit()).isEqualTo("MPa");
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void rawArchiveEnvelopeCarriesLogicalArchiveUri() {
|
||||
VehicleEvent.RawArchive raw = new VehicleEvent.RawArchive(
|
||||
"raw-808-1",
|
||||
"VIN-JT808-001",
|
||||
ProtocolId.JT808,
|
||||
Instant.parse("2026-06-22T08:00:00Z"),
|
||||
Instant.parse("2026-06-22T08:00:01Z"),
|
||||
"trace-raw-1",
|
||||
Map.of(),
|
||||
0x0200,
|
||||
0,
|
||||
new byte[]{0x7e, 0x02, 0x00, 0x7e});
|
||||
|
||||
VehicleEnvelope envelope = new EnvelopeMapper("node-1").toEnvelope(raw);
|
||||
|
||||
String key = RawArchiveKeys.key(raw);
|
||||
assertThat(envelope.hasRawArchive()).isTrue();
|
||||
assertThat(envelope.getRawArchive().getUri()).isEqualTo(RawArchiveKeys.logicalUri(key));
|
||||
assertThat(envelope.getRawArchive().getSizeBytes()).isEqualTo(4);
|
||||
assertThat(envelope.getMetadataMap()).containsEntry(RawArchiveKeys.META_URI, RawArchiveKeys.logicalUri(key));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
package com.lingniu.ingest.sink.mq;
|
||||
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
|
||||
import org.apache.kafka.clients.consumer.ConsumerConfig;
|
||||
import org.apache.kafka.clients.consumer.MockConsumer;
|
||||
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Properties;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
class KafkaEnvelopeConsumerFactoryTest {
|
||||
|
||||
@Test
|
||||
void createsDefaultIndependentConsumersForKnownServiceProcessors() {
|
||||
List<Properties> created = new ArrayList<>();
|
||||
KafkaEnvelopeConsumerFactory factory = new KafkaEnvelopeConsumerFactory(props -> {
|
||||
created.add(props);
|
||||
return new MockConsumer<>(OffsetResetStrategy.EARLIEST);
|
||||
});
|
||||
SinkMqProperties props = new SinkMqProperties();
|
||||
props.setBootstrapServers("kafka-1:9092");
|
||||
EnvelopeConsumerProcessor stateProcessor = processor();
|
||||
EnvelopeConsumerProcessor statProcessor = processor();
|
||||
|
||||
List<KafkaEnvelopeConsumerWorker> workers = factory.createWorkers(Map.of(
|
||||
"vehicleStateEnvelopeConsumerProcessor", stateProcessor,
|
||||
"vehicleStatEnvelopeConsumerProcessor", statProcessor), props);
|
||||
|
||||
assertThat(workers).hasSize(2);
|
||||
assertThat(created)
|
||||
.extracting(p -> p.getProperty(ConsumerConfig.GROUP_ID_CONFIG))
|
||||
.containsExactly("vehicle-state", "vehicle-stat");
|
||||
assertThat(created)
|
||||
.allSatisfy(p -> {
|
||||
assertThat(p.getProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)).isEqualTo("kafka-1:9092");
|
||||
assertThat(p.get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)).isEqualTo(false);
|
||||
});
|
||||
}
|
||||
|
||||
private EnvelopeConsumerProcessor processor() {
|
||||
return new EnvelopeConsumerProcessor(
|
||||
"service",
|
||||
bytes -> EnvelopeIngestResult.processed("evt-1", "VIN001"),
|
||||
record -> {});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
package com.lingniu.ingest.sink.mq;
|
||||
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeConsumerRecord;
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterRecord;
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord;
|
||||
import org.apache.kafka.clients.consumer.MockConsumer;
|
||||
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
|
||||
import org.apache.kafka.common.TopicPartition;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
class KafkaEnvelopeConsumerWorkerTest {
|
||||
|
||||
@Test
|
||||
void pollsKafkaRecordsAndDispatchesThemToConfiguredTopicProcessor() {
|
||||
MockConsumer<String, byte[]> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
|
||||
TopicPartition partition = new TopicPartition("vehicle.realtime", 0);
|
||||
consumer.assign(List.of(partition));
|
||||
consumer.updateBeginningOffsets(Map.of(partition, 0L));
|
||||
consumer.addRecord(new ConsumerRecord<>("vehicle.realtime", 0, 12L, "VIN001", new byte[]{0x01, 0x02}));
|
||||
AtomicReference<EnvelopeDeadLetterRecord> captured = new AtomicReference<>();
|
||||
EnvelopeConsumerProcessor processor = new EnvelopeConsumerProcessor(
|
||||
"vehicle-state",
|
||||
bytes -> EnvelopeIngestResult.invalid("bad envelope"),
|
||||
captured::set);
|
||||
|
||||
int processed = new KafkaEnvelopeConsumerWorker(
|
||||
consumer, Map.of("vehicle.realtime", processor)).pollOnce(Duration.ZERO);
|
||||
|
||||
assertThat(processed).isEqualTo(1);
|
||||
assertThat(captured.get()).satisfies(record -> {
|
||||
assertThat(record.topic()).isEqualTo("vehicle.realtime");
|
||||
assertThat(record.partition()).isEqualTo(0);
|
||||
assertThat(record.offset()).isEqualTo(12L);
|
||||
assertThat(record.key()).isEqualTo("VIN001");
|
||||
assertThat(record.payload()).containsExactly(0x01, 0x02);
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
package com.lingniu.ingest.sink.mq;
|
||||
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeDeadLetterRecord;
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
|
||||
import org.apache.kafka.clients.producer.MockProducer;
|
||||
import org.apache.kafka.clients.producer.ProducerRecord;
|
||||
import org.apache.kafka.common.serialization.ByteArraySerializer;
|
||||
import org.apache.kafka.common.serialization.StringSerializer;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.time.Instant;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
class KafkaEnvelopeDeadLetterSinkTest {
|
||||
|
||||
@Test
|
||||
void publishesDeadLetterRecordToConfiguredDlqTopicWithDiagnosticHeaders() {
|
||||
MockProducer<String, byte[]> producer = new MockProducer<>(
|
||||
true, new StringSerializer(), new ByteArraySerializer());
|
||||
KafkaEnvelopeDeadLetterSink sink = new KafkaEnvelopeDeadLetterSink(producer, "vehicle.consumer.dlq");
|
||||
EnvelopeDeadLetterRecord record = new EnvelopeDeadLetterRecord(
|
||||
"event-history",
|
||||
"vehicle.realtime",
|
||||
3,
|
||||
99L,
|
||||
"VIN001",
|
||||
EnvelopeIngestResult.Status.INVALID_ENVELOPE,
|
||||
"",
|
||||
"",
|
||||
"VehicleEnvelope bytes are invalid",
|
||||
new byte[]{0x01, 0x02},
|
||||
Instant.parse("2026-06-22T08:00:00Z"));
|
||||
|
||||
sink.publish(record);
|
||||
|
||||
assertThat(producer.history()).hasSize(1);
|
||||
ProducerRecord<String, byte[]> sent = producer.history().getFirst();
|
||||
assertThat(sent.topic()).isEqualTo("vehicle.consumer.dlq");
|
||||
assertThat(sent.key()).isEqualTo("VIN001");
|
||||
assertThat(sent.value()).containsExactly(0x01, 0x02);
|
||||
assertThat(header(sent, "dlq-service")).isEqualTo("event-history");
|
||||
assertThat(header(sent, "dlq-source-topic")).isEqualTo("vehicle.realtime");
|
||||
assertThat(header(sent, "dlq-source-partition")).isEqualTo("3");
|
||||
assertThat(header(sent, "dlq-source-offset")).isEqualTo("99");
|
||||
assertThat(header(sent, "dlq-status")).isEqualTo("INVALID_ENVELOPE");
|
||||
assertThat(header(sent, "dlq-message")).isEqualTo("VehicleEnvelope bytes are invalid");
|
||||
}
|
||||
|
||||
private static String header(ProducerRecord<String, byte[]> record, String key) {
|
||||
return new String(record.headers().lastHeader(key).value(), StandardCharsets.UTF_8);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
package com.lingniu.ingest.sink.mq;
|
||||
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeConsumerProcessor;
|
||||
import com.lingniu.ingest.api.consumer.EnvelopeIngestResult;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
class SinkMqConsumerAutoConfigurationTest {
|
||||
|
||||
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
|
||||
.withUserConfiguration(SinkMqAutoConfiguration.class)
|
||||
.withBean("vehicleStateEnvelopeConsumerProcessor", EnvelopeConsumerProcessor.class,
|
||||
() -> new EnvelopeConsumerProcessor(
|
||||
"vehicle-state",
|
||||
bytes -> EnvelopeIngestResult.processed("evt-1", "VIN001"),
|
||||
record -> {}))
|
||||
.withPropertyValues(
|
||||
"lingniu.ingest.sink.mq.enabled=true",
|
||||
"lingniu.ingest.sink.mq.type=kafka",
|
||||
"lingniu.ingest.sink.mq.consumer.enabled=true",
|
||||
"lingniu.ingest.sink.mq.consumer.auto-startup=false",
|
||||
"lingniu.ingest.sink.mq.consumer.bindings.vehicleStateEnvelopeConsumerProcessor.group-id=vehicle-state",
|
||||
"lingniu.ingest.sink.mq.consumer.bindings.vehicleStateEnvelopeConsumerProcessor.topics[0]=vehicle.realtime");
|
||||
|
||||
@Test
|
||||
void createsKafkaEnvelopeConsumerRunnerWhenConsumerBindingIsConfigured() {
|
||||
contextRunner.run(context -> {
|
||||
assertThat(context).hasSingleBean(KafkaEnvelopeConsumerRunner.class);
|
||||
assertThat(context.getBean(KafkaEnvelopeConsumerRunner.class).workers()).hasSize(1);
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
package com.lingniu.ingest.sink.mq;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
class SinkMqPropertiesTest {
|
||||
|
||||
@Test
|
||||
void defaultKafkaBrokerUsesProductionAddressButCanStillBeOverriddenByConfigBinding() {
|
||||
SinkMqProperties props = new SinkMqProperties();
|
||||
|
||||
assertThat(props.getBootstrapServers()).isEqualTo("114.55.58.251:9092");
|
||||
|
||||
props.setBootstrapServers("kafka.internal:9092");
|
||||
assertThat(props.getBootstrapServers()).isEqualTo("kafka.internal:9092");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user