Skip to content

[RFC][Connectors]: Add native s2 input and output connector support in Feldera #5728

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

Open
Mrhs121 wants to merge 9 commits into
base: main
Choose a base branch
from
Open

[RFC][Connectors]: Add native s2 input and output connector support in Feldera #5728

Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
35b3e6a
support s2 native connector
Mrhs121 Mar 2, 2026
3956c0d
support insert_delete for s2 connector
Mrhs121 Mar 2, 2026
8ec5634
support retry connect for s2 connector
Mrhs121 Mar 2, 2026
f853b23
replace manual retry logic with S2 SDK built-in RetryConfig
Mrhs121 Mar 6, 2026
b34d580
support s2_output for s2 connector
Mrhs121 Mar 12, 2026
35388a2
fix s2_output buffer size limit to respect S2 1MB per-record constraint
Mrhs121 Mar 12, 2026
215f527
downgrade s2 input per-record log from info to trace
Mrhs121 Mar 18, 2026
1c18523
move raw format envelope mismatch check to error path to avoid double…
Mrhs121 Mar 18, 2026
feedd68
upgrade s2-sdk from 0.24.4 to 0.26.0 to fix internal dependency conflict
Mrhs121 Mar 23, 2026
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
Prev Previous commit
Next Next commit
downgrade s2 input per-record log from info to trace
  • Loading branch information
@Mrhs121
Mrhs121 committed Mar 23, 2026
commit 215f5273f5b61a168ec99e365929091d05bb7e6d
4 changes: 2 additions & 2 deletions crates/adapters/src/transport/s2/input.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ use tokio::{
task::JoinHandle,
};
use tokio_util::sync::CancellationToken;
use tracing::{Instrument, error, info, info_span};
use tracing::{Instrument, error, info, info_span, trace};
use xxhash_rust::xxh3::Xxh3Default;

#[derive(Debug, Serialize, Deserialize)]
Expand Down Expand Up @@ -368,7 +368,7 @@ fn spawn_s2_reader(
Some(Ok(batch)) => {
// Successfully received a batch
for record in &batch.records {
info!("Got record #{}", record.seq_num);
trace!("Got record #{}", record.seq_num);
next_seq.store(record.seq_num + 1, Ordering::Release);
Copy link
Copy Markdown

@mythical-fred mythical-fred Mar 2, 2026

Choose a reason for hiding this comment

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

Per-record info! will spam logs at any meaningful throughput. Downgrade to trace!, or drop it — the queue/extend logs already give sufficient visibility.

let data = &record.body;
queue.push_with_aux(parser.parse(data, None), Utc::now(), record.seq_num);
Expand Down