Oversized allocation: 299008 bytes in storage::kvstore::flush_and_apply_ops

[ballard26]

Artifact URL: https://ci-artifacts.dev.vectorized.cloud:443/vtools/7520/01880eff-2a13-4369-b9c9-30fc45d70c74/vbuild/output/ducktape-release-clang-amd64.tgz

Test details:

tmp/ducktape/results/2023-05-12--001/LargeControllerSnapshotTest/test_join_restart/220/RedpandaService-0-139958544347200/ip-172-31-13-104/redpanda.log:483

Backtrace:

 [shard 0] seastar_memory - oversized allocation: 299008 bytes. This is non-fatal, but could lead to latency and/or fragmentation issues. Please report: at                   
[Backtrace #1]                                                                                                                                                                                
{tmp/libexec/redpanda} 0x68d9814: void seastar::backtrace<seastar::current_backtrace_tasklocal()::$_3>(seastar::current_backtrace_tasklocal()::$_3&&) at /v/build/v_deps_build/seastar-prefix/
src/seastar/include/seastar/util/backtrace.hh:60                                                                                                                                              
 (inlined by) seastar::current_backtrace_tasklocal() at /v/build/v_deps_build/seastar-prefix/src/seastar/src/util/backtrace.cc:86                                                             
 (inlined by) seastar::current_tasktrace() at /v/build/v_deps_build/seastar-prefix/src/seastar/src/util/backtrace.cc:137                                                                      
 (inlined by) seastar::current_backtrace() at /v/build/v_deps_build/seastar-prefix/src/seastar/src/util/backtrace.cc:170                                                                      
{tmp/libexec/redpanda} 0x65a0581: seastar::memory::cpu_pages::warn_large_allocation(unsigned long) at /v/build/v_deps_build/seastar-prefix/src/seastar/src/core/memory.cc:780                 
 (inlined by) seastar::memory::cpu_pages::check_large_allocation(unsigned long) at /v/build/v_deps_build/seastar-prefix/src/seastar/src/core/memory.cc:788                                    
 (inlined by) seastar::memory::cpu_pages::allocate_large(unsigned int) at /v/build/v_deps_build/seastar-prefix/src/seastar/src/core/memory.cc:794                                             
 (inlined by) seastar::memory::allocate_large(unsigned long) at /v/build/v_deps_build/seastar-prefix/src/seastar/src/core/memory.cc:1345                                                      
 (inlined by) seastar::memory::allocate(unsigned long) at /v/build/v_deps_build/seastar-prefix/src/seastar/src/core/memory.cc:1406                                                            
{tmp/libexec/redpanda} 0x65b0441: operator new(unsigned long) at /v/build/v_deps_build/seastar-prefix/src/seastar/src/core/memory.cc:2087                                                     
{tmp/libexec/redpanda} 0x5a0097a: void* std::__1::__libcpp_operator_new<unsigned long>(unsigned long) at /vectorized/llvm/bin/../include/c++/v1/new:245                                       
 (inlined by) std::__1::__libcpp_allocate(unsigned long, unsigned long) at /vectorized/llvm/bin/../include/c++/v1/new:271                                                                     
 (inlined by) std::__1::allocator<absl::lts_20220623::container_internal::AlignedType<8ul>>::allocate(unsigned long) at /vectorized/llvm/bin/../include/c++/v1/__memory/allocator.h:105       
 (inlined by) std::__1::allocator_traits<std::__1::allocator<absl::lts_20220623::container_internal::AlignedType<8ul>>>::allocate(std::__1::allocator<absl::lts_20220623::container_internal::
AlignedType<8ul>>&, unsigned long) at /vectorized/llvm/bin/../include/c++/v1/__memory/allocator_traits.h:262                                                                                  
 (inlined by) void* absl::lts_20220623::container_internal::Allocate<8ul, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>>(s
td::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>*, unsigned long) at /vectorized/include/absl/container/internal/container_mem
ory.h:64
 (inlined by) absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, 
iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>>::initialize_slots() at /vectorized/
include/absl/container/internal/raw_hash_set.h:1937
 (inlined by) absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, 
iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>>::resize(unsigned long) at /vectoriz
ed/include/absl/container/internal/raw_hash_set.h:1978
 (inlined by) absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, 
iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>>::rehash_and_grow_if_necessary() at 
/vectorized/include/absl/container/internal/raw_hash_set.h:2126
{tmp/libexec/redpanda} 0x5a008da: absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsigne
d int, 31u, false>, iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>>::prepare_insert
(unsigned long) at /vectorized/include/absl/container/internal/raw_hash_set.h:2191
{tmp/libexec/redpanda} 0x59e31d2: std::__1::pair<unsigned long, bool> absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::
basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, fa
lse> const, iobuf>>>::find_or_prepare_insert<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>>(seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const&) at /vect
orized/include/absl/container/internal/raw_hash_set.h:2180
(inlined by) std::__1::pair<absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsigned int
, 31u, false>, iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>>::iterator, bool> abs
l::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>, bytes_typ
e_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>>::EmplaceDecomposable::operator()<seastar::basic_sstr
ing<unsigned char, unsigned int, 31u, false>, std::__1::piecewise_construct_t const&, std::__1::tuple<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>&&>, std::__1::tuple<iobu
f&&>>(seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const&, std::__1::piecewise_construct_t const&, std::__1::tuple<seastar::basic_sstring<unsigned char, unsigned int, 31u,
 false>&&>&&, std::__1::tuple<iobuf&&>&&) const at /vectorized/include/absl/container/internal/raw_hash_set.h:1863
 (inlined by) decltype(std::declval<absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsig
ned int, 31u, false>, iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>>::EmplaceDecom
posable>()(std::declval<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>&& const&>(), std::__1::piecewise_construct, std::declval<std::__1::tuple<seastar::basic_sstring<unsign
ed char, unsigned int, 31u, false>&&>>(), std::declval<std::__1::tuple<iobuf&&>>())) absl::lts_20220623::container_internal::memory_internal::DecomposePairImpl<absl::lts_20220623::container_
internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>, bytes_type_hash, bytes_type_eq, std:
:__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>>::EmplaceDecomposable, seastar::basic_sstring<unsigned char, unsigned int, 31u, 
false>&&, std::__1::tuple<iobuf&&>>(absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsig
ned int, 31u, false>, iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>>::EmplaceDecom
posable&&, std::__1::pair<std::__1::tuple<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>&&>, std::__1::tuple<iobuf&&>>) at /vectorized/include/absl/container/internal/contai
ner_memory.h:139
 (inlined by) decltype(memory_internal::DecomposePairImpl(std::forward<absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar:
:basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, f
alse> const, iobuf>>>::EmplaceDecomposable>(fp), PairArgs(std::forward<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>>(fp0), std::forward<iobuf>(fp0)))) absl::lts_20220623::
container_internal::DecomposePair<absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsigne
d int, 31u, false>, iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>>::EmplaceDecompo
sable, seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>(absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<
seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int
, 31u, false> const, iobuf>>>::EmplaceDecomposable&&, seastar::basic_sstring<unsigned char, unsigned int, 31u, false>&&, iobuf&&) at /vectorized/include/absl/container/internal/container_mem
ory.h:206
 (inlined by) decltype(absl::container_internal::DecomposePair(std::declval<absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<sea
star::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 3
1u, false> const, iobuf>>>::EmplaceDecomposable>(), std::declval<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>>(), std::declval<iobuf>())) absl::lts_20220623::container_int
ernal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>::apply<absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_int
ernal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<un
signed char, unsigned int, 31u, false> const, iobuf>>>::EmplaceDecomposable, seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>(absl::lts_20220623::container_internal::r
aw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>, bytes_type_hash, bytes_type_eq, std::__1::alloc
ator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>>::EmplaceDecomposable&&, seastar::basic_sstring<unsigned char, unsigned int, 31u, false>&&,
 iobuf&&) at /vectorized/include/absl/container/node_hash_map.h:579
 (inlined by) decltype(absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>::apply(std::forward<absl::lts_2022062
3::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>, bytes_type_hash, bytes_
type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>>::EmplaceDecomposable>(fp), std::forward<seastar::basic_sstring<uns
igned char, unsigned int, 31u, false>>(fp0), std::forward<iobuf>(fp0))) absl::lts_20220623::container_internal::hash_policy_traits<absl::lts_20220623::container_internal::NodeHashMapPolicy<s
eastar::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>, void>::apply<absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapP
olicy<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsign
ed int, 31u, false> const, iobuf>>>::EmplaceDecomposable, seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf, absl::lts_20220623::container_internal::NodeHashMapPolicy<se
astar::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>>(absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar:
:basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, f
alse> const, iobuf>>>::EmplaceDecomposable&&, seastar::basic_sstring<unsigned char, unsigned int, 31u, false>&&, iobuf&&) at /vectorized/include/absl/container/internal/hash_policy_traits.h:
170
 (inlined by) std::__1::pair<absl::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsigned int
, 31u, false>, iobuf>, bytes_type_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>>::iterator, bool> abs
l::lts_20220623::container_internal::raw_hash_set<absl::lts_20220623::container_internal::NodeHashMapPolicy<seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, iobuf>, bytes_typ
e_hash, bytes_type_eq, std::__1::allocator<std::__1::pair<seastar::basic_sstring<unsigned char, unsigned int, 31u, false> const, iobuf>>>::emplace<seastar::basic_sstring<unsigned char, unsig
ned int, 31u, false>, iobuf, 0>(seastar::basic_sstring<unsigned char, unsigned int, 31u, false>&&, iobuf&&) at /vectorized/include/absl/container/internal/raw_hash_set.h:1485
 (inlined by) storage::kvstore::apply_op(seastar::basic_sstring<unsigned char, unsigned int, 31u, false>, std::__1::optional<iobuf>) at /var/lib/buildkite-agent/builds/buildkite-amd64-builde
rs-i-06879a8c75b58e683-1/redpanda/redpanda/src/v/storage/kvstore.cc:210
{tmp/libexec/redpanda} 0x59f5cca: storage::kvstore::flush_and_apply_ops()::$_5::operator()() at /var/lib/buildkite-agent/builds/buildkite-amd64-builders-i-06879a8c75b58e683-1/redpanda/redpan
da/src/v/storage/kvstore.cc:255
 (inlined by) decltype(static_cast<storage::kvstore::flush_and_apply_ops()::$_5&>(fp)()) std::__1::__invoke<storage::kvstore::flush_and_apply_ops()::$_5&>(storage::kvstore::flush_and_apply_o
ps()::$_5&) at /vectorized/llvm/bin/../include/c++/v1/type_traits:3640
 (inlined by) std::__1::invoke_result<storage::kvstore::flush_and_apply_ops()::$_5&>::type std::__1::invoke<storage::kvstore::flush_and_apply_ops()::$_5&>(storage::kvstore::flush_and_apply_o
ps()::$_5&) at /vectorized/llvm/bin/../include/c++/v1/__functional/invoke.h:93
 (inlined by) auto seastar::internal::future_invoke<storage::kvstore::flush_and_apply_ops()::$_5&, seastar::internal::monostate>(storage::kvstore::flush_and_apply_ops()::$_5&, seastar::inter
nal::monostate&&) at /vectorized/include/seastar/core/future.hh:1153
 (inlined by) seastar::future<void> seastar::future<void>::then_impl_nrvo<storage::kvstore::flush_and_apply_ops()::$_5, seastar::future<void>>(storage::kvstore::flush_and_apply_ops()::$_5&&)
::'lambda'(seastar::internal::promise_base_with_type<void>&&, storage::kvstore::flush_and_apply_ops()::$_5&, seastar::future_state<seastar::internal::monostate>&&)::operator()(seastar::inter
nal::promise_base_with_type<void>&&, storage::kvstore::flush_and_apply_ops()::$_5&, seastar::future_state<seastar::internal::monostate>&&) const::'lambda'()::operator()() const at /vectorize
d/include/seastar/core/future.hh:1455
 (inlined by) void seastar::futurize<void>::satisfy_with_result_of<seastar::future<void> seastar::future<void>::then_impl_nrvo<storage::kvstore::flush_and_apply_ops()::$_5, seastar::future<v
oid>>(storage::kvstore::flush_and_apply_ops()::$_5&&)::'lambda'(seastar::internal::promise_base_with_type<void>&&, storage::kvstore::flush_and_apply_ops()::$_5&, seastar::future_state<seasta
r::internal::monostate>&&)::operator()(seastar::internal::promise_base_with_type<void>&&, storage::kvstore::flush_and_apply_ops()::$_5&, seastar::future_state<seastar::internal::monostate>&&
) const::'lambda'()>(seastar::internal::promise_base_with_type<void>&&, storage::kvstore::flush_and_apply_ops()::$_5&&) at /vectorized/include/seastar/core/future.hh:1981
 (inlined by) seastar::future<void> seastar::future<void>::then_impl_nrvo<storage::kvstore::flush_and_apply_ops()::$_5, seastar::future<void>>(storage::kvstore::flush_and_apply_ops()::$_5&&)
::'lambda'(seastar::internal::promise_base_with_type<void>&&, storage::kvstore::flush_and_apply_ops()::$_5&, seastar::future_state<seastar::internal::monostate>&&)::operator()(seastar::inter
nal::promise_base_with_type<void>&&, storage::kvstore::flush_and_apply_ops()::$_5&, seastar::future_state<seastar::internal::monostate>&&) const at /vectorized/include/seastar/core/future.hh
:1451
 (inlined by) seastar::continuation<seastar::internal::promise_base_with_type<void>, storage::kvstore::flush_and_apply_ops()::$_5, seastar::future<void> seastar::future<void>::then_impl_nrvo
<storage::kvstore::flush_and_apply_ops()::$_5, seastar::future<void>>(storage::kvstore::flush_and_apply_ops()::$_5&&)::'lambda'(seastar::internal::promise_base_with_type<void>&&, storage::kv
store::flush_and_apply_ops()::$_5&, seastar::future_state<seastar::internal::monostate>&&), void>::run_and_dispose() at /vectorized/include/seastar/core/future.hh:742
{tmp/libexec/redpanda} 0x665b9df: seastar::reactor::run_tasks(seastar::reactor::task_queue&) at /v/build/v_deps_build/seastar-prefix/src/seastar/src/core/reactor.cc:2557
 (inlined by) seastar::reactor::run_some_tasks() at /v/build/v_deps_build/seastar-prefix/src/seastar/src/core/reactor.cc:3020
{tmp/libexec/redpanda} 0x665f657: seastar::reactor::do_run() at /v/build/v_deps_build/seastar-prefix/src/seastar/src/core/reactor.cc:3189
{tmp/libexec/redpanda} 0x665ca29: seastar::reactor::run() at /v/build/v_deps_build/seastar-prefix/src/seastar/src/core/reactor.cc:3072
{tmp/libexec/redpanda} 0x6572411: seastar::app_template::run_deprecated(int, char**, std::__1::function<void ()>&&) at /v/build/v_deps_build/seastar-prefix/src/seastar/src/core/app-template.
cc:265
{tmp/libexec/redpanda} 0x657052f: seastar::app_template::run(int, char**, std::__1::function<seastar::future<int> ()>&&) at /v/build/v_deps_build/seastar-prefix/src/seastar/src/core/app-temp
late.cc:156
{tmp/libexec/redpanda} 0x20aeebe: application::run(int, char**) at /var/lib/buildkite-agent/builds/buildkite-amd64-builders-i-06879a8c75b58e683-1/redpanda/redpanda/src/v/redpanda/application
.cc:349
{tmp/libexec/redpanda} 0x698e3b9: main at /var/lib/buildkite-agent/builds/buildkite-amd64-builders-i-06879a8c75b58e683-1/redpanda/redpanda/src/v/redpanda/main.cc:22
{/opt/redpanda/lib/libc.so.6} 0x2d58f: ?? at ??:0
{/opt/redpanda/lib/libc.so.6} 0x2d648: ?? at ??:0
{tmp/libexec/redpanda} 0x20a9324: _start at ??:0

JIRA Link: CORE-1304

[jcsp]

Hmm, I guess even though we're using node_hash_map here, the sheer number of keys is sufficient to make the array of hash slots an oversized allocation. Maybe we have to use btree_map instead.

Also, flush_and_apply_ops could have a similar batch builder issue to #10749, when many records are written within the flush interval, but that would be a different case to this stack trarce.

We are kind of patching low level issues here without addressing a larger architectural issue, that kvstore was added as this "cheap" utility, but it is actually rather expensive to add keys here, as the footprint is multiplied by the peak partition count per shard. We could simplify our lives somewhat if we made the keys a little bit smarter, to group all the content for a particular NTP together.

[dotnwat]

We could simplify our lives somewhat if we made the keys a little bit smarter, to group all the content for a particular NTP together.

yeh, something that might work in the short term in an transparent-ish way is to make the flat_map be multi-level. first level is i don't know the sub-system key, then some prefix of the next part of the key. not sure this works, or if we should just add an ntp-aware interface.

[BenPope]

Also, flush_and_apply_ops could have a similar batch builder issue to #10749, when many records are written within the flush interval, but that would be a different case to this stack trarce.

I've fixed the above in #11837 but I'm going to unassign myself from this particular fix.

[BenPope]

This isn't fixed by #11837