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

Avoid reordering events due to split partition queues #2437

Merged
merged 5 commits into from
Dec 19, 2024
Merged

Avoid reordering events due to split partition queues #2437

Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
c50d360
Avoid reordering events due to split partition queues
jackkleeman Dec 17, 2024
699aad2
Commit consumer state on rebalance
jackkleeman Dec 17, 2024
49e1f5c
Review comments
jackkleeman Dec 18, 2024
5381ef5
Use spawn_unmanaged
jackkleeman Dec 19, 2024
e1d1926
Install prometheus recorder earlier so kafka metrics work
jackkleeman Dec 19, 2024
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
22 changes: 7 additions & 15 deletions crates/ingress-kafka/src/consumer_task.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ use tokio::sync::{mpsc, oneshot};
use tracing::{debug, info, info_span, warn, Instrument};
use tracing_opentelemetry::OpenTelemetrySpanExt;

use restate_core::{TaskCenter, TaskId, TaskKind};
use restate_core::{TaskCenter, TaskHandle, TaskKind};
use restate_ingress_dispatcher::{
DeduplicationId, DispatchIngressRequest, IngressDispatcher, IngressDispatcherRequest,
};
Expand Down Expand Up @@ -366,19 +366,13 @@ impl ConsumerContext for RebalanceContext {
self.failures_tx.clone(),
);

if let Ok(task_id) = self.task_center.spawn_child(
TaskKind::Ingress, // this task kind exits on error, but we cannot ever return an error, so it doesnt matter.
if let Ok(task_handle) = self.task_center.spawn_unmanaged(
TaskKind::Ingress,
"kafka-partition-ingest",
None,
async {
task.await;
Ok(())
},
task,
) {
topic_partition_tasks.insert(
partition,
AbortOnDrop(self.task_center.clone(), task_id),
);
topic_partition_tasks.insert(partition, AbortOnDrop(task_handle));
} else {
// shutting down
return;
Expand Down Expand Up @@ -418,13 +412,11 @@ impl ConsumerContext for RebalanceContext {
}
}

struct AbortOnDrop(TaskCenter, TaskId);
struct AbortOnDrop(TaskHandle<()>);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Optional: This is a handy type if you want to move to task_center's so others can also use it.


impl Drop for AbortOnDrop {
fn drop(&mut self) {
if let Some(handle) = self.0.cancel_task(self.1) {
handle.abort();
}
self.0.abort();
}
}

Expand Down
Loading