Skip to content
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

Jump to bottom

Fixed Messages Paging #3955

Closed
wants to merge 1 commit into from
Closed

Fixed Messages Paging #3955

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
4 changes: 3 additions & 1 deletion kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/MessagesController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,7 @@ public Mono<ResponseEntity<Flux<TopicMessageEventDTO>>> getTopicMessages(String
SeekTypeDTO seekType,
List<String> seekTo,
Integer limit,
Integer page,
String q,
MessageFilterTypeDTO filterQueryType,
SeekDirectionDTO seekDirection,
Expand All @@ -85,6 +86,7 @@ public Mono<ResponseEntity<Flux<TopicMessageEventDTO>>> getTopicMessages(String
.topicActions(MESSAGES_READ)
.build());

page = page != null ? page : 0;
seekType = seekType != null ? seekType : SeekTypeDTO.BEGINNING;
seekDirection = seekDirection != null ? seekDirection : SeekDirectionDTO.FORWARD;
filterQueryType = filterQueryType != null ? filterQueryType : MessageFilterTypeDTO.STRING_CONTAINS;
Expand All @@ -98,7 +100,7 @@ public Mono<ResponseEntity<Flux<TopicMessageEventDTO>>> getTopicMessages(String
ResponseEntity.ok(
messagesService.loadMessages(
getCluster(clusterName), topicName, positions, q, filterQueryType,
limit, seekDirection, keySerde, valueSerde)
limit, page, seekDirection, keySerde, valueSerde)
)
);

Expand Down
4 changes: 4 additions & 0 deletions kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/AbstractEmitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,4 +59,8 @@ protected void sendFinishStatsAndCompleteSink(FluxSink<TopicMessageEventDTO> sin
messagesProcessing.sendFinishEvent(sink);
sink.complete();
}

protected Integer getPageOffset() {
return messagesProcessing.getPageOffset();
}
}
4 changes: 3 additions & 1 deletion kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/BackwardRecordEmitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,9 @@ public void accept(FluxSink<TopicMessageEventDTO> sink) {
return; //fast return in case of sink cancellation
}
long beginOffset = seekOperations.getBeginOffsets().get(tp);
long readFromOffset = Math.max(beginOffset, readToOffset - msgsToPollPerPartition);
long readFromOffset = Math.max(beginOffset, readToOffset - msgsToPollPerPartition - this.getPageOffset());
readToOffset = readToOffset - this.getPageOffset();


partitionPollIteration(tp, readFromOffset, readToOffset, consumer, sink)
.forEach(r -> sendMessage(sink, r));
Expand Down
4 changes: 4 additions & 0 deletions kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/ForwardRecordEmitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,10 @@ public void accept(FluxSink<TopicMessageEventDTO> sink) {
var seekOperations = SeekOperations.create(consumer, position);
seekOperations.assignAndSeekNonEmptyPartitions();

seekOperations.getOffsetsForSeek().forEach((p, o) -> {
consumer.seek(p, o + getPageOffset());
});

EmptyPollsCounter emptyPolls = pollingSettings.createEmptyPollsCounter();
while (!sink.isCancelled()
&& !sendLimitReached()
Expand Down
10 changes: 9 additions & 1 deletion kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/MessagesProcessing.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,18 +23,26 @@ public class MessagesProcessing {
private final Predicate<TopicMessageDTO> filter;
private final @Nullable Integer limit;

private final @Nullable Integer page;

public MessagesProcessing(ConsumerRecordDeserializer deserializer,
Predicate<TopicMessageDTO> filter,
@Nullable Integer limit) {
@Nullable Integer limit,
@Nullable Integer page) {
this.deserializer = deserializer;
this.filter = filter;
this.limit = limit;
this.page = page;
}

boolean limitReached() {
return limit != null && sentMessages >= limit;
}

Integer getPageOffset() {
return this.limit * this.page;
}

void sendMsg(FluxSink<TopicMessageEventDTO> sink, ConsumerRecord<Bytes, Bytes> rec) {
if (!sink.isCancelled() && !limitReached()) {
TopicMessageDTO topicMessage = deserializer.deserialize(rec);
Expand Down
8 changes: 6 additions & 2 deletions kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/MessagesService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,14 +160,16 @@ public Flux<TopicMessageEventDTO> loadMessages(KafkaCluster cluster, String topi
@Nullable String query,
MessageFilterTypeDTO filterQueryType,
@Nullable Integer pageSize,
Integer page,
SeekDirectionDTO seekDirection,
@Nullable String keySerde,
@Nullable String valueSerde) {
return withExistingTopic(cluster, topic)
.flux()
.publishOn(Schedulers.boundedElastic())
.flatMap(td -> loadMessagesImpl(cluster, topic, consumerPosition, query,
filterQueryType, fixPageSize(pageSize), seekDirection, keySerde, valueSerde));
filterQueryType, fixPageSize(pageSize), page,
seekDirection, keySerde, valueSerde));
}

private int fixPageSize(@Nullable Integer pageSize) {
Expand All @@ -182,6 +184,7 @@ private Flux<TopicMessageEventDTO> loadMessagesImpl(KafkaCluster cluster,
@Nullable String query,
MessageFilterTypeDTO filterQueryType,
int limit,
int page,
SeekDirectionDTO seekDirection,
@Nullable String keySerde,
@Nullable String valueSerde) {
Expand All @@ -191,7 +194,8 @@ private Flux<TopicMessageEventDTO> loadMessagesImpl(KafkaCluster cluster,
var processing = new MessagesProcessing(
deserializationService.deserializerFor(cluster, topic, keySerde, valueSerde),
getMsgFilter(query, filterQueryType),
seekDirection == SeekDirectionDTO.TAILING ? null : limit
seekDirection == SeekDirectionDTO.TAILING ? null : limit,
page
);

if (seekDirection.equals(SeekDirectionDTO.FORWARD)) {
Expand Down
1 change: 1 addition & 0 deletions kafka-ui-api/src/test/java/com/provectus/kafka/ui/emitter/TailingEmitterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,7 @@ private Flux<TopicMessageEventDTO> createTailingFlux(
query,
MessageFilterTypeDTO.STRING_CONTAINS,
0,
0,
SeekDirectionDTO.TAILING,
"String",
"String");
Expand Down
3 changes: 2 additions & 1 deletion kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/MessagesServiceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ void sendMessageReturnsExceptionWhenTopicNotFound() {
@Test
void loadMessagesReturnsExceptionWhenTopicNotFound() {
StepVerifier.create(messagesService
.loadMessages(cluster, NON_EXISTING_TOPIC, null, null, null, 1, null, "String", "String"))
.loadMessages(cluster, NON_EXISTING_TOPIC, null, null, null, 1, 0, null, "String", "String"))
.expectError(TopicNotFoundException.class)
.verify();
}
Expand All @@ -75,6 +75,7 @@ void maskingAppliedOnConfiguredClusters() throws Exception {
null,
null,
100,
0,
SeekDirectionDTO.FORWARD,
StringSerde.name(),
StringSerde.name()
Expand Down
1 change: 1 addition & 0 deletions kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/SendAndReadTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -508,6 +508,7 @@ public void doAssert(Consumer<TopicMessageDTO> msgAssert) {
null,
null,
1,
0,
SeekDirectionDTO.FORWARD,
msgToSend.getKeySerde().get(),
msgToSend.getValueSerde().get()
Expand Down
4 changes: 4 additions & 0 deletions kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -658,6 +658,10 @@ paths:
in: query
schema:
type: integer
- name: page
in: query
schema:
type: integer
- name: q
in: query
schema:
Expand Down
Loading