Upgrade Guava library from 13.0.1 to 32.0.0-jre

[AbhishekKumar9984]

Migrated all removed and deprecated Guava APIs across 74 files:

Objects.toStringHelper() -> MoreObjects.toStringHelper()
Objects.hashCode/equal -> java.util.Objects.hash/equals
new Stopwatch()/elapsedTime() -> Stopwatch.createStarted()/elapsed()
Charsets.UTF_8 -> StandardCharsets.UTF_8
InputSupplier/OutputSupplier -> Callable<Reader/Writer>
Guava Optional -> java.util.Optional
Futures.addCallback/transform 2-arg -> 3-arg with Executor
Futures.immediateCheckedFuture -> Futures.immediateFuture
Service.start/stop/startAndWait/stopAndWait -> startAsync/stopAsync/awaitRunning/awaitTerminated
ServiceListenerAdapter: implements -> extends Service.Listener
Ranges/DiscreteDomains -> Range/DiscreteDomain/ContiguousSet
Files.createTempDir() -> Files.createTempDirectory()
hasher.putString/hashString -> added Charset parameter
Files.newOutputStreamSupplier/newReaderSupplier -> Java NIO equivalents
getServiceName() -> serviceName()

[gemini-code-assist]

Code Review

This pull request upgrades Guava to version 32.0.0-jre and migrates the codebase from deprecated Guava Service lifecycle methods (such as startAndWait() and stopAndWait()) to the newer asynchronous APIs (startAsync(), awaitRunning(), stopAsync(), and awaitTerminated()). It also replaces deprecated Guava utilities with standard Java or updated Guava alternatives, such as using StandardCharsets and java.util.Objects. Feedback on the changes highlights potential hanging issues in Services.java and AbstractTwillService.java due to Guava 15+ not replaying state transitions for listeners added to services that are already in a terminal or target state, and suggests checking the service state prior to registering listeners.

[gemini-code-assist]

In Guava 15+, Service.addListener does not replay state transitions that have already occurred. If a service is already in the target state (e.g., RUNNING when starting, or TERMINATED when stopping), the listener's running() or terminated() callback will never be invoked. This causes the SettableFuture returned by chainStart or chainStop to hang indefinitely.\n\nTo fix this, we should check the current state of the service before calling startAsync() or stopAsync(). If the service is already in the target state, we can immediately complete or proceed to the next service in the chain. Additionally, we should avoid calling startAsync() or stopAsync() if the service is already in a transiting state (like STARTING or STOPPING) to prevent IllegalStateException while still allowing the listener to catch the final transition.

  private static void startOrStop(final boolean doStart, Service service, final Service[] moreServices,\n                                  final AtomicInteger idx,\n                                  final SettableFuture<Service.State> resultFuture) {\n    final java.util.concurrent.atomic.AtomicBoolean completed = new java.util.concurrent.atomic.AtomicBoolean(false);\n\n    service.addListener(new ServiceListenerAdapter() {\n      @Override\n      public void running() {\n        if (doStart && completed.compareAndSet(false, true)) {\n          onServiceOperationComplete(Service.State.RUNNING);\n        }\n      }\n\n      @Override\n      public void terminated(Service.State from) {\n        if (!doStart && completed.compareAndSet(false, true)) {\n          onServiceOperationComplete(Service.State.TERMINATED);\n        }\n      }\n\n      @Override\n      public void failed(Service.State from, Throwable failure) {\n        if (completed.compareAndSet(false, true)) {\n          resultFuture.setException(failure);\n        }\n      }\n\n      private void onServiceOperationComplete(Service.State state) {\n        int nextIdx = idx.getAndIncrement();\n        if (nextIdx >= moreServices.length) {\n          resultFuture.set(state);\n          return;\n        }\n        startOrStop(doStart, moreServices[nextIdx], moreServices, idx, resultFuture);\n      }\n    }, Threads.SAME_THREAD_EXECUTOR);\n\n    Service.State state = service.state();\n    if (doStart) {\n      if (state == Service.State.RUNNING) {\n        if (completed.compareAndSet(false, true)) {\n          int nextIdx = idx.getAndIncrement();\n          if (nextIdx >= moreServices.length) {\n            resultFuture.set(Service.State.RUNNING);\n          } else {\n            startOrStop(doStart, moreServices[nextIdx], moreServices, idx, resultFuture);\n          }\n        }\n        return;\n      } else if (state == Service.State.FAILED) {\n        if (completed.compareAndSet(false, true)) {\n          resultFuture.setException(service.failureCause());\n        }\n        return;\n      } else if (state == Service.State.STOPPING || state == Service.State.TERMINATED) {\n        if (completed.compareAndSet(false, true)) {\n          resultFuture.setException(new IllegalStateException("Cannot start service " + service + " from state " + state));\n        }\n        return;\n      } else if (state == Service.State.NEW) {\n        service.startAsync();\n      }\n    } else {\n      if (state == Service.State.TERMINATED) {\n        if (completed.compareAndSet(false, true)) {\n          int nextIdx = idx.getAndIncrement();\n          if (nextIdx >= moreServices.length) {\n            resultFuture.set(Service.State.TERMINATED);\n          } else {\n            startOrStop(doStart, moreServices[nextIdx], moreServices, idx, resultFuture);\n          }\n        }\n        return;\n      } else if (state == Service.State.FAILED) {\n        if (completed.compareAndSet(false, true)) {\n          resultFuture.setException(service.failureCause());\n        }\n        return;\n      } else if (state == Service.State.NEW || state == Service.State.STARTING || state == Service.State.RUNNING) {\n        service.stopAsync();\n      }\n    }\n  }

[gemini-code-assist]

If the service is already in TERMINATED or FAILED state when the STOP command is received, the listener's terminated() or failed() callback will not be triggered because Guava's addListener does not replay past state transitions. This will prevent messageRemover.run() from being executed, leaving the stop message in ZooKeeper.\n\nTo prevent this, we should check the current state of the service first and execute messageRemover.run() immediately if the service is already terminated or failed.

    State state = state();\n    if (state == State.TERMINATED || state == State.FAILED) {\n      messageRemover.run();\n      return true;\n    }\n\n    addListener(new ServiceListenerAdapter() {\n      @Override\n      public void terminated(State from) {\n        messageRemover.run();\n      }\n\n      @Override\n      public void failed(State from, Throwable failure) {\n        LOG.error("Stop service failed upon STOP command", failure);\n        messageRemover.run();\n      }\n    }, Threads.SAME_THREAD_EXECUTOR);\n    stopAsync();