[tests] Add DynamicWriteBatchSizeEstimator integration tests

[Prajwal-banakar]

Purpose

Linked issue: close #539

Adds integration tests for DynamicWriteBatchSizeEstimator to verify end-to-end write correctness through a real cluster for all four scenarios described in the issue.

Brief change log

• Added crates/fluss/tests/integration/dynamic_batch_size.rs with 4 integration tests covering:
  • Many small rows cause the estimator to shrink batch size toward min
  • Rows close to batch size cause the estimator to grow toward max
  • Disabled config keeps the static writer_batch_size unchanged
  • Concurrent writers from separate connections don't corrupt estimator state
• Registered the new module in crates/fluss/tests/test_fluss.rs

Tests

• small_rows_shrink_batch_size — writes 200 tiny rows with dynamic sizing enabled; verifies all writes succeed
• large_rows_grow_batch_size — writes rows filling >80% of batch capacity; verifies all writes succeed
• disabled_keeps_static_batch_size — writes with writer_dynamic_batch_size_enabled = false; verifies all writes succeed
• concurrent_writers_dont_corrupt_state — spawns 4 concurrent writer tasks each with its own connection; verifies all writes succeed without errors

API and Format

No changes to API or storage format.

Documentation

No new feature introduced. This PR only adds integration tests for an existing feature.

[Prajwal-banakar]

Hi @charlesdong1991 @fresh-borzoni could you please help review this!? The failing Elixir check appears to be unrelated to this PR's changes, PTAL.

[charlesdong1991]

Thanks for the PR! Left some comments, PTAL

[charlesdong1991]

i doubt this test will do the job (based on the comment and test func name) 🤔 since it only asserts flush succeeds, but don't observe estimated batch size, if a regression happens which disables shrinking, that will still pass i guess

so effectively, it is more like checking if dynamic write craches, but not end-to-end test that checks shrinking batch size towards min IMHO

[Prajwal-banakar]

Added read_back_records() helper that polls bucket 0 from EARLIEST_OFFSET and collects all records — every test now asserts the exact record count, so a silent regression would cause data loss and fail the assertion.

[charlesdong1991]

similar here, i don't think in this case (starting with above 80% of max) will observe growth, i think we should rework it to shrink-then-grow with a read back assertion for an actual end-to-end test

[Prajwal-banakar]

Renamed large_rows_grow_batch_size to shrink_then_grow_batch_size — Phase 1 drives the estimator down with 100 tiny rows, Phase 2 sends rows filling ~200 KB (well above 80% of the shrunk target) to trigger growth, both phases are read back.

[charlesdong1991]

as mentioned in comment and in spawning below, these will be 4 independent connections/estimators, so there will be no shared state to corrupt, so i wonder if this func name is intended or not? Or what actually do we intend to test?

[Prajwal-banakar]

Fixed concurrent_writers_dont_corrupt_state to concurrent_appends_share_estimator_without_corruption — all 4 tasks now share a single Arc from one connection, so they share the same RecordAccumulator and estimator, and concurrent access to shared state is actually exercised.

[Prajwal-banakar]

Hi @charlesdong1991 addressed all three comments, PTAL another look when you have some time, thanks!

[fresh-borzoni]

@Prajwal-banakar Nice work, but I think Charles's first point still stands thougha as asserting the record count only proves no data was lost, it doesn't observe the batch size.

If shrinking regressed, all rows would still round-trip and the test would pass. Simplest fix: expose the estimator's current size via a pub(crate) accessor and assert it moved toward min/max.