Introduce global checkpoint listeners

[jasontedor]

This commit introduces the ability for global checkpoint listeners to be registered at the shard level. These listeners are notified when the global checkpoint is updated, and also when the shard closes. To encapsulate these listeners, we introduce a shard-level component that handles synchronization of notification and modifications to the collection of listeners.

Relates #32651

[elasticmachine]

Pinging @elastic/es-distributed

[bleskes]

Thanks @jasontedor . I think we should change the API a bit (see my comment). I also wonder if you considered sharing code between this component and RefreshListeners which is very similar in nature. I say consider as it may very well just end up with generic/boxing hell (as the underlying primitives are different).

[bleskes]

I wonder if we should have parameter that indicates the global checkpoint last sampled by the component trying to register the listener. We can then immediately call the listener if the last global checkpoint this component was notified about (needs to be captured) is higher. It think this would help avoiding race conditions.

[jasontedor]

In my POC I was doing this on the API layer (transport layer) where we have a request with the last known global checkpoint indeed but I agree it makes sense to move that to here.

[jasontedor]

@bleskes It was the boxing/raw generics that I wanted to avoid indeed.

[bleskes]

Nit: I wonder if we should have an onFailure method here for all kind of failures and send the IndexShardClosedException down that route. The down side is of course that people wouldn't be able to pass a method references, but the method won't need to start with if (e != null) etc.

[jasontedor]

I prefer to use a functional interface for enabling the use of lambda expressions.

[bleskes]

Thx @jasontedor . I left some more comments.

[bleskes]

Can we throw an AlreadyClosedException like everywhere else?

[bleskes]

Scratch that. I think we should be consistent and pass IndexShardClosedException to the listener in that case (like it was registered and then we were closed)

[jasontedor]

I pushed 50a9a6c.

[bleskes]

the code runs the listener via the executor, which isn't inline with what the comment says. Which one do you intend to hold?

[jasontedor]

Yeah, the doc went out of sync with the intent. I pushed 22d13a8.

[bleskes]

I think this has to be synchronized to avoid race conditions with add. Say an add method already sampled the lastKnownGlobalCheckpoint and concluded that it has to add a listener. It then goes on to update the listeners array but haven't done so yet (it's still null). The notifyListeners is called here and it runs the listeners != null test before the add method made it non null. It that case I think we miss the listener.

I may be missing something here, but my point is that this is too complex IMO. I don't see why we can't just make this a simple fully synchronized method. The listeners are call on executor anyway.

[jasontedor]

Indeed. I pushed 22d13a8.

[jasontedor]

@ywelsch Would you take over reviewing this PR?

[ywelsch]

Left an initial question

[ywelsch]

the add method assumes that the global checkpoint is always strictly increasing. Add an assertion here that this is so?

[jasontedor]

I pushed 48089fb.

[ywelsch]

This looks as if the listeners are only notified once and then need to reregister if they want more events? I can see why this kind of behavior makes sense for the refresh listeners, with refresh being an expensive operation that ensures that all events registered before the refresh will now see the changes they're waiting for. With global checkpoints, it's less clear to me, as they can be potentially updated many many times per second, so wouldn't you want to stay registered to receive events. If not, will this lead to a storm of reregister events? I think I need to better understand the integration point here, i.e., how this API will be used.

[jasontedor]

@ywelsch See #32651. The intended usage is in CCR where a remote cluster will, when the remote cluster is fully caught up, (remotely) attach a single-use listener to the local cluster for the next global checkpoint change. When the global checkpoint is updated, the listener will be invoked which will return a response to the remote cluster that will act as letting the remote cluster know that there are now additional changes to be fetched.

[ywelsch]

Ok, given this behavior, I wonder if we can make the listener interface simpler (same as Boaz's concern). As the listeners are only notified once and then need to reregister if they want more events, I wonder if it's simpler to just signal an UNASSIGNED_SEQ_NO (or the lastKnownGlobalCheckpoint) on a closing and then have the caller fail on a repeated call to the add method (by throwing directly the exception in that method, not relaying it to the listener). This means that the listener can remain a functional interface (but just a LongConsumer). WDYT?

[jasontedor]

@ywelsch Personally I do not buy the argument that

        @Override
        protected void asyncShardOperation(
                final Request request, final ShardId shardId, final ActionListener<Response> listener) throws IOException {
            final IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
            final IndexShard indexShard = indexService.getShard(shardId.id());
            indexShard.addGlobalCheckpointListener(
                    request.getGlobalCheckpoint(),
                    (g, e) -> {
                        if (g != UNASSIGNED_SEQ_NO) {
                            listener.onResponse(new Response(g));
                        } else {
                            listener.onFailure(e);
                        }
            });
        }

is less clean than

        @Override
        protected void asyncShardOperation(
                final Request request, final ShardId shardId, final ActionListener<Response> listener) throws IOException {
            final IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
            final IndexShard indexShard = indexService.getShard(shardId.id());
            indexShard.addGlobalCheckpointListener(
                    request.getGlobalCheckpoint(),
                    g -> {
                        if (g != UNASSIGNED_SEQ_NO) {
                            listener.onResponse(new Response(g));
                        } else {
                            listener.onFailure(new IndexShardClosedException(shardId));
                        }
            });
        }

We still have to have a check in one form or the other whether or not closing has been signaled to us, so there's always going to be an if check. At least, that is what I read the comment from @bleskes as arguing:

The down side is of course that people wouldn't be able to pass a method references, but the method won't need to start with if (e != null) etc.

In fact, I would argue the approach I have taken is cleaner as it's the shard telling us that we are closed rather than it being signaled indirectly through the value of the global checkpoint passed in the callback. Sure the actual interface is simpler but I prefer the explicit approach. So I think we should either stick with what I have, or have an onClosed callback and lose the ability for the interface to be a functional interface.

Regarding throwing on attempting to register a listener on a closed shard, that was my preferred approach too but @bleskes thought otherwise.

[ywelsch]

There also still the option of adding an onClosed method to the interface with a default NOOP implementation. It will remain a functional interface, and if most tests don't care about the shard closed case, they can treat it like a functional interface. I'll leave the decision to you.

[jasontedor]

It would not be a functional interface for the purposes of the production use-case for this that I have in mind (we have to handle the shard closed event). Thanks, I will leave as-is.

@ywelsch is taking over the review

[ywelsch]

I've left two smaller questions around testing and one ask around simplifying synchronization. Looks good otherwise

[ywelsch]

can you mention here that the listener will be deregistered on notification?

[jasontedor]

I pushed 5b63507.

[ywelsch]

for simplicity, let's move this under the mutex everywhere (and add assert Thread.holdsLock(this) to the beginning of the method). It looks like whenever we call this method, we already go under a mutex for a quick operation before this. In case noone is using the listener functionality, this will therefore amount to the same overhead. In case this infrastructure will be used, listeners should not fly in by the millions in a second, so listeners will mostly be null and there should be no overhead. Let's optimize this in the future if we see any issue. I was scratching my head for a bit to check if concurrency was correct here (and I believe it is), but it's not worth the complexity imho.

[jasontedor]

Sure. I pushed 88dee76.

[ywelsch]

Can you also add closing to this test? So that each listener gets notified exactly once, even under concurrent closing.

[jasontedor]

Good idea, this caught an issue. I pushed 9533679.

[ywelsch]

why 3 here? How do we ensure we end up running every thread to completion? (I have a hard time counting the barrier.await() here)

[jasontedor]

There are three threads. The thread repeatedly firing updates, the thread repeatedly adding listeners, and the main test thread. We use the barrier to synchronize their start (they all wait on the barrier, then go), and to signal when the updating and adding thread are done (they all wait on the barrier again). So within the updating thread we have:

wait on barrier
do the updating loop
wait on barrier

while within the adding thread we have

wait on barrier
do the adding loop
wait on barrier

and the main test thread does two waits

wait on barrier
wait on barrier

The first wait corresponds to the top wait of the updating and adding threads, and the second wait corresponds to the bottom wait of the updating and adding threads.

Does that help?

[jasontedor]

Maybe it is best to look at it like this:

Updating thread:

wait on barrier
do the updating loop
wait on barrier

Adding thread:

wait on barrier
do the adding loop
wait on barrier

Main test thread:

wait on barrier
no-op
wait on barrier

[ywelsch]

ok, thanks for the explanation, makes sense. Can you add a comment to that effect?

[jasontedor]

@ywelsch I pushed 07dab4f.

[jasontedor]

@ywelsch This is ready for another round from you.

[ywelsch]

LGTM

[ywelsch]

ok, thanks for the explanation, makes sense. Can you add a comment to that effect?