feat(kafka): add message display ordering

This commit is contained in:
Guoyu Su authored and GitHub committed 2026-08-31 23:47:51 +08:00
1 parent 42c0dc7f89
commit 2fe26052a7
12 files changed
+171 -26

No files matched your search

@@ -7,6 +7,7 @@ import { mqPeekMessages } from "@/lib/backend/api";
import { formatError } from "@/lib/backend/errorUtils";
import { copyToClipboard } from "@/lib/common/clipboard";
import { buildKafkaMessageSearchText, kafkaMessageSearchTextMatches, normalizeKafkaMessageSearchQuery } from "@/lib/mq/kafkaMessageSearch";
import { sortKafkaMessagesByPublishTime, type KafkaMessageDisplayOrder } from "@/lib/mq/kafkaMessageSort";
import { parseNonNegativeSafeInteger } from "@/lib/mq/mqPeekFilters";
import { useToast } from "@/composables/useToast";
import { Button } from "@/components/ui/button";
@@ -40,6 +41,7 @@ const advancedExpanded = ref(false);
const messageSearchQuery = ref("");
type KafkaPeekStartPosition = NonNullable<PeekMessagesOptions["startPosition"]>;
const kafkaStartPosition = ref<KafkaPeekStartPosition>("latest");
const kafkaMessageDisplayOrder = ref<KafkaMessageDisplayOrder>("newest");
let messageRequestVersion = 0;
const isKafka = computed(() => props.mqSystemKind === "kafka");
@@ -57,6 +59,7 @@ const filteredMessages = computed(() => {
if (!query) return messages.value;
return searchableMessages.value.filter(({ searchText }) => kafkaMessageSearchTextMatches(searchText, query)).map(({ message }) => message);
});
const displayedMessages = computed(() => (isKafka.value ? sortKafkaMessagesByPublishTime(filteredMessages.value, kafkaMessageDisplayOrder.value) : filteredMessages.value));
function peekGroupName(): string {
if (props.mqSystemKind === "rocketmq") return "__dbx_rocketmq_viewer__";
@@ -223,6 +226,18 @@ watch(kafkaStartPosition, () => {
</SelectContent>
</Select>
</label>
<label v-if="isKafka">
<span>{{ t("mqMessages.timeOrder") }}</span>
<Select v-model="kafkaMessageDisplayOrder">
<SelectTrigger data-testid="kafka-message-display-order" class="message-browser-start-position">
<SelectValue />
</SelectTrigger>
<SelectContent position="popper" class="message-browser-start-position-content">
<SelectItem value="newest">{{ t("mqMessages.newestFirst") }}</SelectItem>
<SelectItem value="oldest">{{ t("mqMessages.oldestFirst") }}</SelectItem>
</SelectContent>
</Select>
</label>
<label v-if="isKafka">
<span>{{ t("mqMessages.partition") }}</span>
<input v-model="partition" data-testid="kafka-peek-partition" type="number" min="0" :placeholder="t('mqMessages.partitionPlaceholderAll')" :disabled="loading" />
@@ -263,7 +278,7 @@ watch(kafkaStartPosition, () => {
<div v-else-if="!messages.length" class="message-empty">{{ t("mqMessages.noMessages") }}</div>
<div v-else-if="!filteredMessages.length" class="message-empty" data-testid="kafka-message-filter-empty">{{ t("mqMessages.noMatchingMessages") }}</div>
<div v-else class="message-list">
<article v-for="message in filteredMessages" :key="`${message.properties?.partition ?? 'p'}-${message.messageId || message.position}`" class="message-row">
<article v-for="message in displayedMessages" :key="`${message.properties?.partition ?? 'p'}-${message.messageId || message.position}`" class="message-row">
<div class="message-meta">
<span>#{{ message.position }}</span>
<span v-if="message.properties?.partition != null">{{ t("mqMessages.metaPartition", { partition: message.properties.partition }) }}</span>
@@ -74,13 +74,14 @@ async function setSelectValue(trigger: HTMLElement, value: string) {
await flushUi();
}
async function mountBrowser(mqSystemKind: "kafka" | "rabbitmq" = "kafka") {
async function mountBrowser(mqSystemKind: "kafka" | "rabbitmq" = "kafka", appearance: "form" | "monitoring" = "form") {
root = document.createElement("div");
document.body.appendChild(root);
app = createApp(MessageBrowser, {
connectionId: "mq-1",
topic: { ...TOPIC, tenant: mqSystemKind === "rabbitmq" ? "_rabbitmq" : "_kafka" },
mqSystemKind,
appearance,
});
app.mount(root);
await flushUi();
@@ -121,6 +122,10 @@ async function loadMessages(container: ParentNode) {
await flushUi();
}
function displayedPayloads(container: ParentNode): string[] {
return [...container.querySelectorAll<HTMLElement>(".message-payload")].map((element) => element.textContent || "");
}
beforeEach(() => {
backend.mqPeekMessages.mockReset();
clipboard.copyToClipboard.mockReset();
@@ -229,6 +234,51 @@ describe("MessageBrowser", () => {
expect(backend.mqPeekMessages).toHaveBeenCalledWith("mq-1", expect.objectContaining({ topic: "events" }), "__dbx_kafka_viewer__", 20, { startPosition: "latest" });
});
it("orders loaded Kafka messages newest first by default and changes presentation without reloading", async () => {
backend.mqPeekMessages.mockResolvedValueOnce([
{ position: 1, messageId: "1", publishTime: "1000", payloadBase64: "", payloadText: "oldest", properties: {}, headers: {} },
{ position: 2, messageId: "2", publishTime: "3000", payloadBase64: "", payloadText: "newest", properties: {}, headers: {} },
{ position: 3, messageId: "3", publishTime: "2000", payloadBase64: "", payloadText: "middle", properties: {}, headers: {} },
]);
const browser = await mountBrowser();
await loadMessages(browser);
expect(displayedPayloads(browser)).toEqual(["newest", "middle", "oldest"]);
const displayOrder = browser.querySelector<HTMLElement>('[data-testid="kafka-message-display-order"]');
if (!displayOrder) throw new Error("Kafka message display order select not found");
await setSelectValue(displayOrder, "oldest");
expect(displayedPayloads(browser)).toEqual(["oldest", "middle", "newest"]);
expect(backend.mqPeekMessages).toHaveBeenCalledTimes(1);
});
it("keeps the selected Kafka display order when filtering loaded messages", async () => {
backend.mqPeekMessages.mockResolvedValueOnce([
{ position: 1, messageId: "1", publishTime: "1000", payloadBase64: "", payloadText: "matching oldest", properties: {}, headers: {} },
{ position: 2, messageId: "2", publishTime: "3000", payloadBase64: "", payloadText: "matching newest", properties: {}, headers: {} },
{ position: 3, messageId: "3", publishTime: "2000", payloadBase64: "", payloadText: "excluded", properties: {}, headers: {} },
]);
const browser = await mountBrowser();
await loadMessages(browser);
const displayOrder = browser.querySelector<HTMLElement>('[data-testid="kafka-message-display-order"]');
const filter = browser.querySelector<HTMLInputElement>('[data-testid="kafka-message-filter-input"]');
if (!displayOrder || !filter) throw new Error("Kafka message presentation controls not found");
await setSelectValue(displayOrder, "oldest");
await setInputValue(filter, "matching");
expect(displayedPayloads(browser)).toEqual(["matching oldest", "matching newest"]);
expect(backend.mqPeekMessages).toHaveBeenCalledTimes(1);
});
it("shows Kafka display ordering in the monitoring appearance", async () => {
const browser = await mountBrowser("kafka", "monitoring");
expect(browser.querySelector('[data-testid="kafka-message-display-order"]')).not.toBeNull();
expect(browser.querySelector('[data-testid="kafka-peek-start-position"]')).toBeNull();
});
it("normalizes a decimal count before sending the request", async () => {
const browser = await mountBrowser();
const countInput = browser.querySelector<HTMLInputElement>('[data-testid="peek-count"]');
@@ -383,6 +433,7 @@ describe("MessageBrowser", () => {
await loadMessages(browser);
expect(browser.querySelector('[data-testid="kafka-message-filter"]')).toBeNull();
expect(browser.querySelector('[data-testid="kafka-message-display-order"]')).toBeNull();
});
it("sends explicit earliest and offset read positions", async () => {
+6 -3
View File
@@ -7847,9 +7847,12 @@ export default {
kafkaLatest: "Recent N messages (default)",
kafkaEarliest: "Earliest messages",
kafkaOffset: "From offset",
kafkaLatestHint: "Reads up to N candidates from each partition, then returns the newest N by timestamp across the topic.",
kafkaEarliestHint: "Reads history forward from each selected partition.",
kafkaOffsetHint: "Reads forward from the specified offset in the chosen partition, or every partition when left empty.",
timeOrder: "Time order",
newestFirst: "Newest first",
oldestFirst: "Oldest first",
kafkaLatestHint: "Reads up to N candidates from each partition, then selects the newest N by timestamp across the topic.",
kafkaEarliestHint: "Reads from the earliest available message in each selected partition.",
kafkaOffsetHint: "Reads from the specified offset in the chosen partition, or every partition when left empty.",
messagesLoading: "Loading messages...",
noMessages: "No messages",
filterLoadedPlaceholder: "Filter loaded messages",
+6 -3
View File
@@ -7748,9 +7748,12 @@ export default withEnglishFallback({
kafkaLatest: "Últimos N mensajes (predeterminado)",
kafkaEarliest: "Mensajes más antiguos",
kafkaOffset: "Desde offset",
kafkaLatestHint: "Lee hasta N candidatos por partición y devuelve los N más recientes del tema por marca de tiempo.",
kafkaEarliestHint: "Lee el historial hacia adelante desde cada partición seleccionada.",
kafkaOffsetHint: "Lee hacia adelante desde el offset indicado en la partición elegida, o en todas si se deja vacía.",
timeOrder: "Orden temporal",
newestFirst: "Más recientes primero",
oldestFirst: "Más antiguos primero",
kafkaLatestHint: "Lee hasta N candidatos por partición y selecciona los N más recientes del tema por marca de tiempo.",
kafkaEarliestHint: "Lee desde el primer mensaje disponible de cada partición seleccionada.",
kafkaOffsetHint: "Lee desde el offset indicado en la partición elegida, o en todas si se deja vacía.",
messagesLoading: "Cargando mensajes...",
noMessages: "No hay mensajes",
filterLoadedPlaceholder: "Filtrar mensajes cargados",
+6 -3
View File
@@ -7749,9 +7749,12 @@ export default withEnglishFallback({
kafkaLatest: "Ultimi N messaggi (predefinito)",
kafkaEarliest: "Primi messaggi",
kafkaOffset: "Da offset",
kafkaLatestHint: "Legge fino a N candidati per partizione e restituisce i N messaggi più recenti del topic in base al timestamp.",
kafkaEarliestHint: "Legge la cronologia in avanti da ogni partizione selezionata.",
kafkaOffsetHint: "Legge in avanti dall'offset indicato nella partizione scelta, o in tutte se lasciata vuota.",
timeOrder: "Ordine temporale",
newestFirst: "Più recenti prima",
oldestFirst: "Meno recenti prima",
kafkaLatestHint: "Legge fino a N candidati per partizione e seleziona i N messaggi più recenti del topic in base al timestamp.",
kafkaEarliestHint: "Legge dal primo messaggio disponibile di ogni partizione selezionata.",
kafkaOffsetHint: "Legge dall'offset indicato nella partizione scelta, o in tutte se lasciata vuota.",
messagesLoading: "Caricamento messaggi...",
noMessages: "Nessun messaggio",
filterLoadedPlaceholder: "Filtra i messaggi caricati",
+6 -3
View File
@@ -7800,9 +7800,12 @@ export default withEnglishFallback({
kafkaLatest: "最新 N 件(デフォルト)",
kafkaEarliest: "最も古いメッセージ",
kafkaOffset: "Offset から読み取る",
kafkaLatestHint: "各パーティションから最大 N 件の候補を読み取り、トピック全体でタイムスタンプが新しい N 件を返します。",
kafkaEarliestHint: "選択した各パーティションの先頭から昇順に読み取ります。",
kafkaOffsetHint: "指定 Offset から、選択したパーティション(空欄ならすべて)を昇順に読み取ります。",
timeOrder: "時刻順",
newestFirst: "新しい順",
oldestFirst: "古い順",
kafkaLatestHint: "各パーティションから最大 N 件の候補を読み取り、トピック全体でタイムスタンプが新しい N 件を選択します。",
kafkaEarliestHint: "選択した各パーティションで最も古い利用可能なメッセージから読み取ります。",
kafkaOffsetHint: "指定 Offset から、選択したパーティション(空欄ならすべて)を読み取ります。",
messagesLoading: "メッセージを読み込み中...",
noMessages: "メッセージがありません",
filterLoadedPlaceholder: "読み込み済みメッセージを絞り込む",
+6 -3
View File
@@ -7204,9 +7204,12 @@ export default withEnglishFallback({
kafkaLatest: "최근 N개 메시지(기본값)",
kafkaEarliest: "가장 이른 메시지",
kafkaOffset: "오프셋부터 읽기",
kafkaLatestHint: "파티션마다 최대 N개의 후보를 읽은 뒤 토픽 전체에서 타임스탬프가 가장 최신인 N개를 반환합니다.",
kafkaEarliestHint: "선택한 각 파티션의 처음부터 오름차순으로 읽습니다.",
kafkaOffsetHint: "지정한 오프셋부터 선택한 파티션(비워 두면 전체)을 오름차순으로 읽습니다.",
timeOrder: "시간 순서",
newestFirst: "최신순",
oldestFirst: "오래된순",
kafkaLatestHint: "파티션마다 최대 N개의 후보를 읽은 뒤 토픽 전체에서 타임스탬프가 가장 최신인 N개를 선택합니다.",
kafkaEarliestHint: "선택한 각 파티션에서 가장 이른 메시지부터 읽습니다.",
kafkaOffsetHint: "지정한 오프셋부터 선택한 파티션(비워 두면 전체)을 읽습니다.",
messagesLoading: "메시지를 불러오는 중...",
noMessages: "메시지가 없습니다",
filterLoadedPlaceholder: "불러온 메시지 필터링",
+6 -3
View File
@@ -7750,9 +7750,12 @@ export default withEnglishFallback({
kafkaLatest: "Últimas N mensagens (padrão)",
kafkaEarliest: "Mensagens mais antigas",
kafkaOffset: "A partir do offset",
kafkaLatestHint: "Lê até N candidatos por partição e retorna as N mensagens mais recentes do tópico por data e hora.",
kafkaEarliestHint: "Lê o histórico em ordem crescente de cada partição selecionada.",
kafkaOffsetHint: "Lê em ordem crescente a partir do offset informado na partição escolhida, ou em todas se deixada vazia.",
timeOrder: "Ordem temporal",
newestFirst: "Mais recentes primeiro",
oldestFirst: "Mais antigas primeiro",
kafkaLatestHint: "Lê até N candidatos por partição e seleciona as N mensagens mais recentes do tópico por data e hora.",
kafkaEarliestHint: "Lê a partir da primeira mensagem disponível de cada partição selecionada.",
kafkaOffsetHint: "Lê a partir do offset informado na partição escolhida, ou em todas se deixada vazia.",
messagesLoading: "Carregando mensagens...",
noMessages: "Nenhuma mensagem",
filterLoadedPlaceholder: "Filtrar mensagens carregadas",
+6 -3
View File
@@ -7829,9 +7829,12 @@ export default withEnglishFallback({
kafkaLatest: "最近 N 条(默认)",
kafkaEarliest: "最早消息",
kafkaOffset: "从指定 Offset 读取",
kafkaLatestHint: "每个分区最多读取 N 条候选消息,再按时间全局倒序返回最新 N 条。",
kafkaEarliestHint: "从每个选定分区的最早可读消息开始正序读取。",
kafkaOffsetHint: "从指定 Offset 开始正序读取;填写分区时仅读该分区,留空则对全部分区分别读取。",
timeOrder: "时间顺序",
newestFirst: "最新优先",
oldestFirst: "最早优先",
kafkaLatestHint: "每个分区最多读取 N 条候选消息,再按时间选出全局最新 N 条。",
kafkaEarliestHint: "从每个选定分区的最早可读消息开始读取。",
kafkaOffsetHint: "从指定 Offset 开始读取;填写分区时仅读该分区,留空则对全部分区分别读取。",
messagesLoading: "消息加载中...",
noMessages: "暂无消息",
filterLoadedPlaceholder: "筛选已加载消息",
+6 -3
View File
@@ -7738,9 +7738,12 @@ export default withEnglishFallback({
kafkaLatest: "最近 N 則(預設)",
kafkaEarliest: "最早訊息",
kafkaOffset: "從指定 Offset 讀取",
kafkaLatestHint: "每個分區最多讀取 N 筆候選訊息,再依時間全域倒序回傳最新 N 筆。",
kafkaEarliestHint: "從每個選定分區最早可讀的訊息開始正序讀取。",
kafkaOffsetHint: "從指定 Offset 開始正序讀取;填寫分區時只讀該分區,留空則對全部分區分別讀取。",
timeOrder: "時間順序",
newestFirst: "最新優先",
oldestFirst: "最早優先",
kafkaLatestHint: "每個分區最多讀取 N 筆候選訊息,再依時間選出全域最新 N 筆。",
kafkaEarliestHint: "從每個選定分區最早可讀的訊息開始讀取。",
kafkaOffsetHint: "從指定 Offset 開始讀取;填寫分區時只讀該分區,留空則對全部分區分別讀取。",
messagesLoading: "訊息載入中...",
noMessages: "暫無訊息",
filterLoadedPlaceholder: "篩選已載入訊息",
@@ -0,0 +1,32 @@
import { describe, expect, it } from "vitest";
import type { PeekedMessage } from "@/types/mq";
import { sortKafkaMessagesByPublishTime } from "@/lib/mq/kafkaMessageSort";
function message(position: number, publishTime?: string): PeekedMessage {
return {
position,
messageId: String(position),
publishTime,
payloadBase64: "",
payloadText: `message ${position}`,
properties: {},
headers: {},
};
}
describe("Kafka message display sorting", () => {
it("sorts valid timestamps in either direction without mutating the input", () => {
const messages = [message(1, "1000"), message(2, "3000"), message(3, "2000")];
expect(sortKafkaMessagesByPublishTime(messages, "newest").map(({ position }) => position)).toEqual([2, 3, 1]);
expect(sortKafkaMessagesByPublishTime(messages, "oldest").map(({ position }) => position)).toEqual([1, 3, 2]);
expect(messages.map(({ position }) => position)).toEqual([1, 2, 3]);
});
it("keeps equal and unavailable timestamps stable after valid timestamps", () => {
const messages = [message(1), message(2, "2000"), message(3, "invalid"), message(4, "2000"), message(5, "")];
expect(sortKafkaMessagesByPublishTime(messages, "newest").map(({ position }) => position)).toEqual([2, 4, 1, 3, 5]);
expect(sortKafkaMessagesByPublishTime(messages, "oldest").map(({ position }) => position)).toEqual([2, 4, 1, 3, 5]);
});
});
@@ -0,0 +1,23 @@
import type { PeekedMessage } from "@/types/mq";
export type KafkaMessageDisplayOrder = "newest" | "oldest";
function parsePublishTime(value?: string): number | null {
if (value == null || value.trim() === "") return null;
const timestamp = Number(value);
return Number.isFinite(timestamp) ? timestamp : null;
}
export function sortKafkaMessagesByPublishTime(messages: readonly PeekedMessage[], order: KafkaMessageDisplayOrder): PeekedMessage[] {
return messages
.map((message, index) => ({ message, index, timestamp: parsePublishTime(message.publishTime) }))
.sort((left, right) => {
if (left.timestamp == null && right.timestamp != null) return 1;
if (left.timestamp != null && right.timestamp == null) return -1;
if (left.timestamp != null && right.timestamp != null && left.timestamp !== right.timestamp) {
return order === "newest" ? right.timestamp - left.timestamp : left.timestamp - right.timestamp;
}
return left.index - right.index;
})
.map(({ message }) => message);
}