I don't love this. Can we use a semaphore instead?

I have added a semaphore implementation 0 permits. Please let me know if this impl looks good to you
private final AtomicInteger numOfRpcs = new AtomicInteger(0);

Hmm, semaphore doesn't seem quite right for this. Sorry for misleading you on that

Maybe something like this:

AtomicInteger numOutstandingBatches = new AtomicInteger();
Object flushLock = new Object();

void sendBatch() {
  numOutstandingBatches.increment();
}

void onBatchCompletion() {
   if (numOutstandingBatches.decrementAndGet() == 0) {
   flushLock.notifyAll();
 }
}

void flush() {
  sendBatch();
  awaitAllOutstandingBatches();
}

void awaitAllOutstandingBatches() {
  while(numOutstandingBatches.get() > 0) {
    synchronized(lock) {
      flushLock.wait();
    }
  }
}

Thanks for the detailed steps, I have implemented with the same for Batcher#flush(). Please have a look.

static?

My intention was not to declare type params for inner class... but Have updated it to private static class.

Please add a comment to make that clear

Hmm, semaphore doesn't seem quite right for this. Sorry for misleading you on that

Maybe something like this:

AtomicInteger numOutstandingBatches = new AtomicInteger();
Object flushLock = new Object();

void sendBatch() {
  numOutstandingBatches.increment();
}

void onBatchCompletion() {
   if (numOutstandingBatches.decrementAndGet() == 0) {
   flushLock.notifyAll();
 }
}

void flush() {
  sendBatch();
  awaitAllOutstandingBatches();
}

void awaitAllOutstandingBatches() {
  while(numOutstandingBatches.get() > 0) {
    synchronized(lock) {
      flushLock.wait();
    }
  }
}

I'm not sure I understand why we need a timeout

I added this as a defense, Also inspired by existing CBT-OperationalAccountant

What is it defending against?

I added this for the case of RPC taking too long, which would enable this thread to keep on checking the numOfOutStandingBatch.

I now realized that this would not add any meaning. The only update we are waiting here would be from onBatchComplete's notifyAll(), which will execute on all cases.
Thanks a lot for pointing it out. I will remove this timeout.

Please do not put period (.) at the end of messages like this to stay consistent with the rest of the codebase (applies to other places as well, like line 113).

Also the message probably should not use the class names to indicate what is null (either use field names i.e prototype instead of RequestPrototype, or regular langugae to describe what is null, otherwise it looks like the "class" itself is null).

can this be @Autovalue? I'm not a big fan of @Autovalue but we already use it extensively in gax-java, so lets reuse it here if possible to reduce boilerplate code typing.

I don't consider BatcherImpl is a value object. So i'm not sure how AutoValue can help here

I don't think it has to be a value object, it can be anything, which can be constructed with a builder.
Now when I look at it, do we even need a builder here? I'm just trying to figure out if we can reduce boilerplate code here (either by generating builder with the autovalue annotation or not having it at all).

I dont think a builder is necessary here. Given the choice between AutoValue and removing the builder I would prefer removing the builder

I have removed the Builder from the current change and annotated this class with @AutoValue.

Just a side note, We may need to come back on this when introducing flow controllers. As after implementing triggers, we would be expecting 6 arguments(i.e. existing ones + BatchingSettings, FlowController, and an Executor).

TLDR version: please remove AutoValue for the time being.

Longer version:
I've given this more thought, and I really think that AutoValue is really the wrong tool for this. In the best practices it states

Don't use AutoValue to implement value semantics unless you really want value semantics. In particular, you should never care about the difference between two equal instances.

This is definitely not the case for BatcherImpl since a Batcher has mutable state and we do care which instance we call flush on. Also, using AutoValue here implies that equality will use value semantics and be determined by the prototype, callable & batchingDescriptor:

BatcherImpl.Builder builder = ...;

Batcher b1 = builder.build();
Batcher b2 = builder.build();
b2.add(...)

b1.equals(b2) == true; 

Which makes no sense. I think the only meaningful equality here is referential and goes against the best practices of AutoValue.

Since you already removed the Builder until a later PR, please remove the use of AutoValue here since it buys you very little. Once that PR comes up I'll chat with vam about the best path forward.

Please make it volatile

This is only used by the single caller thread, so it doesn't need to be volatile

Ok, but it is practically free to make it such, ans feels safer (a little bit of defensive programming never hurts in concurrent programming).

I generally prefer to avoid defensive programming because you end up hiding the intention of the code. But I agree with you that volatile is free here. So if you feel strongly, then I won't object

I would still add volatile

@rahulKQL can you add the volatile and I will merge this PR

I am sorry, I haven't marked this ongoing conversation as closed.
I have already added a volatile for isClosed flag.