GeezerVison

The World Wide Web, restored to 800 pixels of glory.

GeezerVison is showing github.com — [connectors] Recover Kafka output transactions through broker outages · feldera/feldera@1ae5241

File tree

Expand file treeCollapse file tree

Open diff view settings

Filter options

crates/connectors/src
- controller.rs
- transport.rs
- transport
  - kafka.rs
  - kafka
    - ft.rs
    - ft
      - output.rs
      - output
        tests.rs
        tests
        proxy.rs
        real.rs
      - spool.rs
      - test.rs
    - metrics.rs
    - nonft.rs
    - nonft
      - output.rs
docs.feldera.com/docs
- changelog.md
- connectors/sinks
  - kafka.md

Expand file treeCollapse file tree

Open diff view settings

Collapse file

`‎crates/connectors/src/controller.rs‎`

Copy file name to clipboardExpand all lines: crates/connectors/src/controller.rs

+1Lines changed: 1 addition & 0 deletions

Original file line number	Diff line number	Diff line change
`@@ -8577,6 +8577,7 @@ impl ControllerInner {`
`8577`	`8577`	`endpoint_name,`
`8578`	`8578`	`self.fault_tolerance == Some(FtModel::ExactlyOnce),`
`8579`	`8579`	`&self.secrets_dir,`
	`8580`	`+ self.storage_path.as_deref(),`
`8580`	`8581`	`self.shutdown_token().clone(),`
`8581`	`8582`	`)`
`8582`	`8583`	`.map_err(\|e\| ControllerError::output_transport_error(endpoint_name, true, e))?;`

Collapse file

`‎crates/connectors/src/transport.rs‎`

Copy file name to clipboardExpand all lines: crates/connectors/src/transport.rs

+7-1Lines changed: 7 additions & 1 deletion

Original file line number	Diff line number	Diff line change
`@@ -195,6 +195,9 @@ pub fn input_transport_config_to_endpoint(`
`195`	`195`	`/// waits on something outside the pipeline uses it to stop waiting when the`
`196`	`196`	`/// pipeline goes down.`
`197`	`197`	`///`
	`198`	+/// `storage_path` is the pipeline's storage directory, if it has one on the
	`199`	`+/// local file system. An endpoint that spills to disk uses it.`
	`200`	`+///`
`198`	`201`	`/// Returns an error if there is a invalid configuration for the endpoint.`
`199`	`202`	/// Returns `None` if the transport configuration variant is incompatible with an output endpoint.
`200`	`203`	`#[allow(unused_variables)]`
`@@ -203,6 +206,7 @@ pub fn output_transport_config_to_endpoint(`
`203`	`206`	`endpoint_name: &str,`
`204`	`207`	`fault_tolerant: bool,`
`205`	`208`	`secrets_dir: &Path,`
	`209`	`+ storage_path: Option<&Path>,`
`206`	`210`	`shutdown: CancellationToken,`
`207`	`211`	`) -> AnyResult<Option<Box<dyn OutputEndpoint>>> {`
`208`	`212`	`let config = resolve_secret_references_via_json(secrets_dir, config)?;`
`@@ -216,7 +220,9 @@ pub fn output_transport_config_to_endpoint(`
`216`	`220`	`shutdown,`
`217`	`221`	`)?))),`
`218`	`222`	`true => Ok(Some(Box::new(KafkaFtOutputEndpoint::new(`
`219`		`- config, shutdown,`
	`223`	`+ config,`
	`224`	`+ shutdown,`
	`225`	`+ storage_path.map(Path::to_owned),`
`220`	`226`	`)?))),`
`221`	`227`	`},`
`222`	`228`	`#[cfg(feature = "with-redis")]`

Collapse file

`‎crates/connectors/src/transport/kafka.rs‎`

Copy file name to clipboardExpand all lines: crates/connectors/src/transport/kafka.rs

+1Lines changed: 1 addition & 0 deletions

Original file line number	Diff line number	Diff line change
`@@ -38,6 +38,7 @@ pub use ft::{KafkaFtInputEndpoint, KafkaFtOutputEndpoint};`
`38`	`38`	`pub use nonft::KafkaOutputEndpoint;`
`39`	`39`
`40`	`40`	`pub(crate) mod ft;`
	`41`	`+pub(crate) mod metrics;`
`41`	`42`	`mod nonft;`
`42`	`43`
`43`	`44`	`const MAX_POLLING_INTERVAL: Duration = Duration::from_millis(5000);`

Collapse file

`‎crates/connectors/src/transport/kafka/ft.rs‎`

Copy file name to clipboardExpand all lines: crates/connectors/src/transport/kafka/ft.rs

+1-1Lines changed: 1 addition & 1 deletion

Original file line number	Diff line number	Diff line change
`@@ -7,6 +7,7 @@`
`7`	`7`	`//! for a step.`
`8`	`8`	`mod input;`
`9`	`9`	`mod output;`
	`10`	`+mod spool;`
`10`	`11`
`11`	`12`	`use anyhow::{Context, Result as AnyResult, bail};`
`12`	`13`	`use feldera_types::transport::kafka::{KafkaLogLevel, default_redpanda_server};`
`@@ -169,7 +170,6 @@ impl CommonConfig {`
`169`	`170`	`("enable.idempotence", "true"),`
`170`	`171`	`("batch.size", "1"),`
`171`	`172`	`("batch.num.messages", "1"),`
`172`		`- ("retries", "5"),`
`173`	`173`	`("socket.nagle.disable", "true"),`
`174`	`174`	`("linger.ms", "0"),`
`175`	`175`	`];`

0 commit comments

Comments

0 (0)

You are visitor 01402520 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.