Skip to content

fix: handle gaps in offsets #2252

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
May 5, 2025
Merged

fix: handle gaps in offsets #2252

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
175 changes: 132 additions & 43 deletions src/containers/Tenant/Diagnostics/TopicData/__test__/getData.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,25 +3,6 @@ import {prepareResponse} from '../getData';
import {TOPIC_DATA_FETCH_LIMIT} from '../utils/constants';

describe('prepareResponse', () => {
// Test case 1: Normal case with no removed messages
test('should handle case with no removed messages', () => {
const response: TopicDataResponse = {
StartOffset: '100',
EndOffset: '120',
Messages: [{Offset: '100'}, {Offset: '101'}, {Offset: '102'}] as TopicMessage[],
};

const result = prepareResponse(response, 100);

expect(result.start).toBe(100);
expect(result.end).toBe(120);
expect(result.messages.length).toBe(3);
expect(result.messages[0]).toEqual({Offset: '100'});
expect(result.messages[1]).toEqual({Offset: '101'});
expect(result.messages[2]).toEqual({Offset: '102'});
});

// Test case 2: Case with some removed messages
test('should handle case with some removed messages', () => {
const response: TopicDataResponse = {
StartOffset: '105',
Expand All @@ -33,22 +14,20 @@ describe('prepareResponse', () => {

expect(result.start).toBe(105);
expect(result.end).toBe(120);
expect(result.messages.length).toBe(8); // 5 removed + 3 actual
expect(result.messages.length).toBe(20);

// Check removed messages
expect(result.messages[0]).toEqual({Offset: '100', removed: true});
expect(result.messages[1]).toEqual({Offset: '101', removed: true});
expect(result.messages[2]).toEqual({Offset: '102', removed: true});
expect(result.messages[3]).toEqual({Offset: '103', removed: true});
expect(result.messages[4]).toEqual({Offset: '104', removed: true});

// Check actual messages
expect(result.messages[0]).toEqual({Offset: 100, removed: true});
expect(result.messages[1]).toEqual({Offset: 101, removed: true});
expect(result.messages[2]).toEqual({Offset: 102, removed: true});
expect(result.messages[3]).toEqual({Offset: 103, removed: true});
expect(result.messages[4]).toEqual({Offset: 104, removed: true});
expect(result.messages[5]).toEqual({Offset: '105'});
expect(result.messages[6]).toEqual({Offset: '106'});
expect(result.messages[7]).toEqual({Offset: '107'});
expect(result.messages[8]).toEqual({Offset: 108, removed: true});
expect(result.messages[19]).toEqual({Offset: 119, removed: true});
});

// Test case 3: Case with more removed messages than the limit
test('should handle case with more removed messages than the limit', () => {
const response: TopicDataResponse = {
StartOffset: '150',
Expand All @@ -64,11 +43,10 @@ describe('prepareResponse', () => {

// All messages should be "removed" placeholders since there are more than the limit
for (let i = 0; i < TOPIC_DATA_FETCH_LIMIT; i++) {
expect(result.messages[i]).toEqual({Offset: `${100 + i}`, removed: true});
expect(result.messages[i]).toEqual({Offset: 100 + i, removed: true});
}
});

// Test case 4: Case with non-numeric offsets
test('should handle case with non-numeric offsets', () => {
const response: TopicDataResponse = {
StartOffset: 'not-a-number',
Expand All @@ -82,33 +60,32 @@ describe('prepareResponse', () => {
expect(result.start).toBe(0);
expect(result.end).toBe(0);

// Since start (0) < offset (100), removedMessagesCount is negative
// No removed messages should be added
expect(result.messages.length).toBe(2);
expect(result.messages[0]).toEqual({Offset: '100'});
expect(result.messages[1]).toEqual({Offset: '101'});
// Since end (0) <= offset (100), no messages should be processed
expect(result.messages.length).toBe(0);
});

// Test case 5: Case with empty Messages array
test('should handle case with empty Messages array', () => {
const response: TopicDataResponse = {
StartOffset: '100',
EndOffset: '100',
EndOffset: '120',
Messages: [],
};

const result = prepareResponse(response, 100);

expect(result.start).toBe(100);
expect(result.end).toBe(100);
expect(result.messages.length).toBe(0);
expect(result.end).toBe(120);
// Should have placeholders for all offsets in range
expect(result.messages.length).toBe(TOPIC_DATA_FETCH_LIMIT);
// All should be marked as removed
expect(result.messages[0]).toEqual({Offset: 100, removed: true});
expect(result.messages[19]).toEqual({Offset: 119, removed: true});
});

// Test case 6: Case with more messages than the limit
test('should handle case with more messages than the limit', () => {
// Create an array of 30 messages (more than TOPIC_DATA_FETCH_LIMIT)
const messages: TopicMessage[] = [];
for (let i = 0; i < TOPIC_DATA_FETCH_LIMIT + 1; i++) {
for (let i = 0; i < TOPIC_DATA_FETCH_LIMIT + 10; i++) {
messages.push({Offset: `${100 + i}`} as TopicMessage);
}

Expand All @@ -132,7 +109,6 @@ describe('prepareResponse', () => {
expect(result.messages[2]).toEqual({Offset: '102'});
});

// Test case 7: Case with both removed messages and actual messages within limit
test('should handle case with both removed and actual messages within limit', () => {
const response: TopicDataResponse = {
StartOffset: '110',
Expand All @@ -147,5 +123,118 @@ describe('prepareResponse', () => {

// 10 removed + 10 actual = 20 (TOPIC_DATA_FETCH_LIMIT)
expect(result.messages.length).toBe(TOPIC_DATA_FETCH_LIMIT);

// Check first 10 messages are marked as removed
for (let i = 0; i < 10; i++) {
expect(result.messages[i]).toEqual({Offset: 100 + i, removed: true});
}

// Check next 10 messages are actual messages
for (let i = 0; i < 10; i++) {
expect(result.messages[i + 10]).toEqual({Offset: `${110 + i}`});
}
});

test('should handle case with gaps in message offsets', () => {
const response: TopicDataResponse = {
StartOffset: '100',
EndOffset: '110',
Messages: [
{Offset: '100'} as TopicMessage,
{Offset: '102'} as TopicMessage,
{Offset: '105'} as TopicMessage,
{Offset: '109'} as TopicMessage,
],
};

const result = prepareResponse(response, 100);

expect(result.start).toBe(100);
expect(result.end).toBe(110);
expect(result.messages.length).toBe(10);

// Check actual messages
expect(result.messages[0]).toEqual({Offset: '100'});
expect(result.messages[2]).toEqual({Offset: '102'});
expect(result.messages[5]).toEqual({Offset: '105'});
expect(result.messages[9]).toEqual({Offset: '109'});

// Check removed messages (gaps)
expect(result.messages[1]).toEqual({Offset: 101, removed: true});
expect(result.messages[3]).toEqual({Offset: 103, removed: true});
expect(result.messages[4]).toEqual({Offset: 104, removed: true});
expect(result.messages[6]).toEqual({Offset: 106, removed: true});
expect(result.messages[7]).toEqual({Offset: 107, removed: true});
expect(result.messages[8]).toEqual({Offset: 108, removed: true});
});

test('should handle case with offset greater than EndOffset', () => {
const response: TopicDataResponse = {
StartOffset: '100',
EndOffset: '110',
Messages: [{Offset: '100'}, {Offset: '101'}, {Offset: '102'}] as TopicMessage[],
};

const result = prepareResponse(response, 115);

expect(result.start).toBe(100);
expect(result.end).toBe(110);
// Since offset > end, no messages should be processed
expect(result.messages.length).toBe(0);
});

test('should handle case with offset equal to EndOffset', () => {
const response: TopicDataResponse = {
StartOffset: '100',
EndOffset: '110',
Messages: [{Offset: '100'}, {Offset: '101'}, {Offset: '102'}] as TopicMessage[],
};

const result = prepareResponse(response, 110);

expect(result.start).toBe(100);
expect(result.end).toBe(110);
// Since offset = end, no messages should be processed
expect(result.messages.length).toBe(0);
});

test('should handle case with undefined Messages', () => {
const response: TopicDataResponse = {
StartOffset: '100',
EndOffset: '110',
// Messages is undefined
};

const result = prepareResponse(response, 100);

expect(result.start).toBe(100);
expect(result.end).toBe(110);
// Should have placeholders for all offsets in range
expect(result.messages.length).toBe(10);
// All should be marked as removed
for (let i = 0; i < 10; i++) {
expect(result.messages[i]).toEqual({Offset: 100 + i, removed: true});
}
});

test('should handle zero StartOffset and EndOffset', () => {
const response: TopicDataResponse = {
StartOffset: '0',
EndOffset: '10',
Messages: [{Offset: '0'}, {Offset: '1'}, {Offset: '2'}] as TopicMessage[],
};

const result = prepareResponse(response, 0);

expect(result.start).toBe(0);
expect(result.end).toBe(10);
// Since offset equals start, only actual messages should be returned
expect(result.messages.length).toBe(10);
// Check if the function correctly handles zero offsets
expect(result.messages[0]).toEqual({Offset: '0'});
expect(result.messages[1]).toEqual({Offset: '1'});
expect(result.messages[2]).toEqual({Offset: '2'});
expect(result.messages[3]).toEqual({Offset: 3, removed: true});
expect(result.messages[9]).toEqual({Offset: 9, removed: true});
});
});
50 changes: 29 additions & 21 deletions src/containers/Tenant/Diagnostics/TopicData/getData.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import {isNil} from 'lodash';

import type {FetchData} from '../../../../components/PaginatedTable';
import {TOPIC_MESSAGE_SIZE_LIMIT} from '../../../../store/reducers/topic';
import type {
TopicDataRequest,
TopicDataResponse,
Expand All @@ -26,28 +27,34 @@ export function prepareResponse(response: TopicDataResponse, offset: number) {
const start = safeParseNumber(StartOffset);
const end = safeParseNumber(EndOffset);

const removedMessagesCount = start - offset;

const result: TopicMessageEnhanced[] = [];
for (let i = 0; i < Math.min(TOPIC_DATA_FETCH_LIMIT, removedMessagesCount); i++) {
result.push({
Offset: String(offset + i),
removed: true,
});
}
for (
let i = 0;
i <
Math.min(
TOPIC_DATA_FETCH_LIMIT,
TOPIC_DATA_FETCH_LIMIT - removedMessagesCount,
Messages.length,
);
i++
) {
result.push(Messages[i]);
const normalizedMessages: TopicMessageEnhanced[] = [];

const limit = Math.min(TOPIC_DATA_FETCH_LIMIT, Math.max(end - offset, 0));
let i = 0;
let j = 0;
while (j < limit) {
const currentOffset = offset + j;
const currentMessage = Messages[i];
if (currentOffset < start) {
normalizedMessages.push({
Offset: currentOffset,
removed: true,
});
} else if (
!isNil(currentMessage?.Offset) &&
String(currentMessage.Offset) === String(currentOffset)
) {
normalizedMessages.push(currentMessage);
i++;
} else {
normalizedMessages.push({
Offset: currentOffset,
removed: true,
});
}
j++;
}
return {start, end, messages: result};
return {start, end, messages: normalizedMessages};
}

export const generateTopicDataGetter = ({
Expand Down Expand Up @@ -77,6 +84,7 @@ export const generateTopicDataGetter = ({
partition,
limit,
last_offset: normalizedOffset + limit,
message_size_limit: TOPIC_MESSAGE_SIZE_LIMIT,
};
queryParams.offset = normalizedOffset;

Expand Down
Loading