flake_id_generator_impl::new_id() diverges from Java AutoBatcher: many parallel batch fetches, INT64_MIN leak, wasted batches, and validity<=0 bug [HZ-5440]

[ihsandemir]

Summary

flake_id_generator_impl::new_id() is intended to mirror the Hazelcast Java
client's batching logic (com.hazelcast.flakeidgen.impl.AutoBatcher#newId(),
as used by ClientFlakeIdGeneratorProxy). Because the C++ client exposes an
async API (new_id() returns boost::future<int64_t>) while the Java
AutoBatcher.newId() is synchronous (it holds synchronized(this) across
a blocking newIdBatch(...).joinInternal() and loops in for(;;) until it has
an ID), the current C++ implementation took shortcuts that diverge from the
reference and cause real defects under concurrency, plus a hard bug for a
valid configuration.

Affected code

• hazelcast/src/hazelcast/client/proxy.cpp
  • flake_id_generator_impl::new_id() — lines 1156–1172
  • flake_id_generator_impl::new_id_batch() — lines 1174–1189
  • flake_id_generator_impl::new_id_internal() — lines 1142–1154
  • flake_id_generator_impl::Block::Block() — lines 1010–1016
  • flake_id_generator_impl::Block::next() — lines 1018–1033
• hazelcast/include/hazelcast/client/proxy/flake_id_generator_impl.h
  • unused std::mutex lock_; — line 146
  • IdBatch / Block are private nested classes — lines 68–137

Reference: com.hazelcast.flakeidgen.impl.AutoBatcher#newId()
(AutoBatcher.java:46-62) and ClientFlakeIdGeneratorProxy
(ClientFlakeIdGeneratorProxy.java:46-56) in the Hazelcast Java repository.

Problems

1. Thundering herd — no single-flight on batch fetch

There is no equivalent of Java's synchronized(this). When N threads call
new_id() concurrently and the current batch is exhausted/expired, all N
threads independently call new_id_batch(), producing N network requests and
N server-reserved ID batches, of which N-1 are discarded. This defeats the
purpose of prefetching and wastes server-side ID space. The
std::mutex lock_; member declared in the header (line 146) is never used,
indicating the synchronization was intended but not implemented.

2. No retry loop — INT64_MIN can be returned to the caller

The .then() continuation calls newBlock->next() exactly once and returns
its result (proxy.cpp:1167-1170). Java instead loops (for(;;)) until it
obtains a real ID. If the freshly fetched block yields nothing (batch size 0,
or already expired by the time the continuation runs), C++ returns INT64_MIN
to the user as if it were a valid ID.

3. Ignored compare_exchange_strong result — wasted/leaked batches

auto value = newBlock->next();          // ID #0 already consumed from newBlock
auto b = block_.load();
block_.compare_exchange_strong(b, newBlock);   // result ignored
return value;

If the CAS fails (another thread installed a different block), newBlock is
silently dropped while value was already taken from it — the rest of that
server-reserved batch is leaked. If the CAS succeeds, it may overwrite a
still-valid fresh block another thread just installed, leaking that batch
instead. Java's double-checked if (block != this.block) continue; prevents
both.

4. validity <= 0 semantics missing (hard bug)

Java Block:

this.invalidSince = validity > 0 ? Clock.currentTimeMillis() + validity : Long.MAX_VALUE;

i.e. a non-positive validity means "never expire". The C++ Block
constructor (proxy.cpp:1013) does:

invalid_since_(std::chrono::steady_clock::now() + validity)

unconditionally. With a Java-valid configuration of prefetch_validity = 0,
every block is born already expired, so Block::next() always returns
INT64_MIN and every new_id() call fails.

5. Testability gap

new_id_batch() is welded to the network (invoke(...)), and IdBatch /
Block are private nested classes. The Java unit tests
(com.hazelcast.flakeidgen.impl.AutoBatcherTest,
FlakeIdGenerator_IdBatchTest) are deterministic because AutoBatcher takes
an injectable IdBatchSupplier ("a separate class due to testability", per its
own Javadoc). There is no equivalent seam in C++, so that coverage cannot be
ported.

Why a literal Java port is not possible

new_id() must keep returning boost::future<int64_t> and its continuation
runs on a Boost.Asio IO thread. It therefore can neither block the caller nor
hold a std::mutex across the batch network round-trip the way Java does.

Proposed fix

Extract a testable auto_batcher (the C++ analogue of Java's AutoBatcher)
and fix new_id() with a single-flight batch fetch whose winner is
offloaded to the user executor — the same pattern reliable_topic already
uses (get_client_execution_service().get_user_executor() +
boost::async(executor, ...)). This also aligns the C++ structure with Java,
where ClientFlakeIdGeneratorProxy holds an AutoBatcher.

New units (proxy namespace):

• id_batch — promote IdBatch/iterator to a standalone, unit-testable class.
• auto_batcher — owns the batching/coalescing state and logic:
  • boost::atomic_shared_ptr<block> block_; (lock-free fast path)
  • std::mutex mutex_; — guards winner election + in-flight handle only,
    never held across the network call
  • boost::shared_future<std::shared_ptr<block>> fetch_in_progress_; +
    uint64_t fetch_generation_ to reset the handle after completion
  • util::hz_thread_pool& executor_; (user executor)
  • std::function<boost::future<id_batch>(int32_t)> batch_supplier_; — the
    injectable seam (Java's IdBatchSupplier); real proxy passes the
    invoke(...)-based lambda, unit tests pass a deterministic fake
  • block keeps the existing lock-free CAS next() plus a new
    non-consuming bool usable() const
• flake_id_generator_impl becomes thin: owns an auto_batcher built with the
  real-fetch lambda + config + user executor; new_id() delegates.

auto_batcher::new_id() algorithm:

1. Lock-free fast path: block_->next(); if valid → make_ready_future.
2. Under mutex_: double-check the block changed (Java's
   if (block != this.block) continue); if no fetch in progress, become the
   winner, bump fetch_generation_, launch the winner on executor_, and
   publish fetch_in_progress_ as a shared_future; copy the handle and
   release the mutex.
3. Winner (on a user-executor thread, blocking is acceptable there):
   for(;;) { batch = batch_supplier_(batch_size_).get(); block_ = make_shared<block>(...); if (usable) { install; return; } }.
   Exceptions propagate through the shared_future to all waiters.
4. Every caller (winner-side and losers) attaches a .then(launch::sync, ...)
   continuation (never blocking the caller): clear the in-flight handle if
   still the same generation; f.get(); block_->next() for a distinct ID via
   the existing CAS; if still INT64_MIN (more concurrent waiters than
   prefetch_count), recurse into new_id() (.unwrap() keeps the
   boost::future<int64_t> signature) — the async analogue of Java's outer
   loop.

Orthogonal bug fixes (independent of the above):

• block ctor: invalid_since_ = validity > milliseconds::zero() ? now() + validity : steady_clock::time_point::max(); (Java <= 0 = never
  expire).
• Add non-consuming block::usable():
  invalid_since_ > now() && num_returned_ < batch_size.

This yields Java behavioral parity: one outstanding batch request per
generator, no thundering herd, no wasted/leaked batches, never returns
INT64_MIN, and the public API stays fully async/non-blocking.

Testing plan

Add tests next to FlakeIdGeneratorApiTest in HazelcastTests1.cpp. The
auto_batcher seam makes the deterministic ones run with no server
(util::hz_thread_pool is standalone-constructible).

Ported from com.hazelcast.flakeidgen.impl:

Java test	C++ port
FlakeIdConcurrencyTestUtil.concurrentlyGenerateIds	helper concurrently_generate_ids(gen, 4, 100000); assert size == threads*ids
AutoBatcherTest.when_validButUsedAll_then_fetchNew	auto_batcher{3, 10000ms, fake}, fake returns id_batch(base,1,size); base+=size; expect 0,1,2,3
AutoBatcherTest.when_notValid_then_fetchNew	same fake, validity 100ms; new_id()→0; sleep 150ms; new_id()→3
AutoBatcherTest.concurrencySmokeTest	helper over auto_batcher w/ fake; assert full coverage [0, threads*ids) and fake call-count == expected #batches (proves single-flight)
FlakeIdGenerator_IdBatchTest.testIterator / _hasNextNotCalled / _emptyIdBatch	id_batch unit tests: id_batch(1,10,2) yields 1,11 then == end(); id_batch(1,10,0) iterator()==end()
FlakeIdGenerator_IdBatchTest.testIterator_hasNextMultipleCalls	adapt/skip — C++ IdIterator is single-pass; Java hasNext() idempotency N/A
FlakeIdGeneratorProxyTest, *BackpressureTest, *MemberIntegrationTest, *NodeIdOverflowIntegrationTest	not ported — member/server-side, outside the client new_id() path

Plus a new validity == 0 regression test (no Java equivalent):
auto_batcher with validity = 0ms; assert repeated new_id().get() over
time never returns INT64_MIN and never throws.

Existing real-server tests (FlakeIdGeneratorApiTest.testStartingValue,
testSmoke, testAddGetFlakeIdGeneratorIntegrity) must stay green. Build
Debug (-fsanitize=address, -Werror) and run the auto_batcher/id_batch
unit tests under ASan to verify no data races on block_ /
fetch_in_progress_ / mutex_.

[github-actions]

Internal Jira issue: API-2381