github-actions[bot] commented on code in PR #4219:
URL: https://github.com/apache/iggy/pull/4219#discussion_r4126836464
##########
core/connectors/sinks/delta_sink/src/lib.rs:
##########
@@ -53,6 +53,7 @@ struct SinkState {
}
#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
Review Comment:
nit: An unknown `plugin_config` key now fails the sink at open, where serde
ignored it before, and the README does not say so. Add one sentence to the
plugin configuration section that states that unknown keys fail initialization.
##########
core/connectors/sinks/delta_sink/src/sink.rs:
##########
@@ -37,33 +37,64 @@ impl Sink for DeltaSink {
);
let table_url = url::Url::parse(&self.config.table_uri).map_err(|e| {
- error!("Failed to parse table URI '{}': {e}",
self.config.table_uri);
- Error::InitError(format!("Invalid table URI: {e}"))
+ error!(
Review Comment:
simplification: The new `open()` error paths write each sentence twice, once
to `error!` and once to the returned `Error`. The parse site's two texts
already differ. Have one helper format the sentence, log it, and return the
matching `Error`. Also at lines 61, 69, 74, 86 and 93.
##########
core/integration/tests/connectors/delta/delta_sink.rs:
##########
@@ -235,3 +238,116 @@ async fn delta_sink_writes_to_s3(harness: &TestHarness,
fixture: DeltaS3Fixture)
assert_eq!(row_count, message_count);
}
+
+#[iggy_harness(
+ server(connectors_runtime(config_path =
"tests/connectors/delta/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn delta_sink_handles_bucket_absence_on_s3(
+ harness: &TestHarness,
+ _fixture: DeltaS3NoBucketFixture,
+) {
+ let runtime = harness
+ .connectors_runtime()
+ .expect("connectors runtime handle should be available");
+ let (stdout, stderr) = runtime.collect_logs();
Review Comment:
nit: With `IGGY_TEST_VERBOSE` set the harness inherits the runtime output,
so `collect_logs` returns `[No stdout log]` and these assertions fail. Skip
them when that variable is set, as
`core/integration/tests/cluster/crash_recovery_corruption.rs:272` does. Also at
lines 272, 290, 308, 328 and 347.
##########
core/connectors/sinks/delta_sink/src/storage.rs:
##########
@@ -27,25 +28,42 @@ pub(crate) fn build_storage_options(
match config.storage_backend_type {
Some(StorageBackendType::S3) => {
- let access_key = config.aws_s3_access_key.as_ref().ok_or_else(|| {
- Error::InitError("S3 backend requires
'aws_s3_access_key'".into())
- })?;
- let secret_key = config.aws_s3_secret_key.as_ref().ok_or_else(|| {
- Error::InitError("S3 backend requires
'aws_s3_secret_key'".into())
+ match (
+ config.aws_s3_access_key.as_ref(),
+ config.aws_s3_secret_key.as_ref(),
+ ) {
+ (Some(access_key), Some(secret_key)) => {
+ info!("S3 backend: using static access key/secret key
credentials");
+ opts.insert(
+ "AWS_ACCESS_KEY_ID".into(),
+ access_key.expose_secret().to_owned(),
+ );
+ opts.insert(
+ "AWS_SECRET_ACCESS_KEY".into(),
+ secret_key.expose_secret().to_owned(),
+ );
+ }
+ (None, None) => {
Review Comment:
nit: Every S3 fixture sets static keys, so the keyless arm added here has no
end-to-end test. Add one S3 test that omits both keys and puts
`AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` in the runtime environment.
##########
core/integration/tests/connectors/fixtures/delta/fixture.rs:
##########
@@ -200,6 +200,68 @@ impl TestFixture for DeltaFixture {
}
}
+pub struct DeltaCorruptedLogFixture {
+ _temp_dir: TempDir,
+ table_path: PathBuf,
+}
+
+impl DeltaCorruptedLogFixture {
+ // Overwrites the version-0 commit file with garbage bytes but keeps its
+ // name intact: `is_delta_table_location` only lists filenames under
+ // `_delta_log`, so the table still "exists"; `DeltaTable::load` is what
+ // then fails parsing this file during log replay.
+ async fn corrupt_commit_log(table_path: &Path) -> Result<(),
TestBinaryError> {
+ let commit_path = table_path
+ .join("_delta_log")
+ .join("00000000000000000000.json");
+ tokio::fs::write(&commit_path, b"not a valid delta log commit entry")
+ .await
+ .map_err(|error| TestBinaryError::FixtureSetup {
+ fixture_type: "DeltaCorruptedLogFixture".to_string(),
+ message: format!(
+ "Failed to corrupt commit log at {}: {error}",
+ commit_path.display()
+ ),
+ })
+ }
+}
+
+#[async_trait]
+impl TestFixture for DeltaCorruptedLogFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let temp_dir = TempDir::new().map_err(|error|
TestBinaryError::FixtureSetup {
+ fixture_type: "DeltaCorruptedLogFixture".to_string(),
+ message: format!("Failed to create temp directory: {error}"),
+ })?;
+
+ let table_path = temp_dir.path().join("delta_table");
+ let table_uri = format!("file://{}", table_path.display());
+ DeltaFixture::create_table(&table_uri).await?;
Review Comment:
simplification: `DeltaCorruptedLogFixture::setup` and its env map repeat
`DeltaFixture`'s code, so the two can drift apart. Hold a `DeltaFixture` here
and corrupt `inner.table_path` from it, as the new S3 fixtures wrap
`DeltaS3Fixture`. Saves: about 30 lines.
##########
core/integration/tests/connectors/fixtures/delta/fixture.rs:
##########
@@ -388,3 +450,87 @@ impl TestFixture for DeltaS3Fixture {
envs
}
}
+
+pub struct DeltaS3NoTableFixture {
+ inner: DeltaS3Fixture,
+}
+
+#[async_trait]
+impl TestFixture for DeltaS3NoTableFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let id = Uuid::new_v4();
+ let network = format!("iggy-delta-s3-{id}");
+ let minio_name =
fixtures::unique_container_name("minio-delta-no-table");
+
+ let (minio, minio_endpoint) = DeltaS3Fixture::start_minio(&network,
&minio_name).await?;
Review Comment:
simplification: `DeltaS3NoTableFixture` and `DeltaS3NoBucketFixture` copy
the same MinIO startup body, and the two copies differ by the container name
and one `create_bucket` call. Add one `DeltaS3Fixture` constructor that starts
MinIO and returns the fixture, then call it from both. Saves: about 25 lines.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
