build(deps): bump grpc from 1.73.0 to 1.74.0

Bumps `grpc` from 1.73.0 to 1.74.0.

Updates `io.grpc:grpc-core` from 1.73.0 to 1.74.0
- [Release notes](https://github.com/grpc/grpc-java/releases)
- [Commits](https://github.com/grpc/grpc-java/compare/v1.73.0...v1.74.0)

Updates `io.grpc:grpc-netty` from 1.73.0 to 1.74.0
- [Release notes](https://github.com/grpc/grpc-java/releases)
- [Commits](https://github.com/grpc/grpc-java/compare/v1.73.0...v1.74.0)

Updates `io.grpc:grpc-protobuf` from 1.73.0 to 1.74.0
- [Release notes](https://github.com/grpc/grpc-java/releases)
- [Commits](https://github.com/grpc/grpc-java/compare/v1.73.0...v1.74.0)

Updates `io.grpc:grpc-stub` from 1.73.0 to 1.74.0
- [Release notes](https://github.com/grpc/grpc-java/releases)
- [Commits](https://github.com/grpc/grpc-java/compare/v1.73.0...v1.74.0)

Updates `io.grpc:grpc-grpclb` from 1.73.0 to 1.74.0
- [Release notes](https://github.com/grpc/grpc-java/releases)
- [Commits](https://github.com/grpc/grpc-java/compare/v1.73.0...v1.74.0)

Updates `io.grpc:grpc-inprocess` from 1.73.0 to 1.74.0
- [Release notes](https://github.com/grpc/grpc-java/releases)
- [Commits](https://github.com/grpc/grpc-java/compare/v1.73.0...v1.74.0)

Updates `io.grpc:grpc-util` from 1.73.0 to 1.74.0
- [Release notes](https://github.com/grpc/grpc-java/releases)
- [Commits](https://github.com/grpc/grpc-java/compare/v1.73.0...v1.74.0)

---
updated-dependencies:
- dependency-name: io.grpc:grpc-core
  dependency-version: 1.74.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: io.grpc:grpc-netty
  dependency-version: 1.74.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: io.grpc:grpc-protobuf
  dependency-version: 1.74.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: io.grpc:grpc-stub
  dependency-version: 1.74.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: io.grpc:grpc-grpclb
  dependency-version: 1.74.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: io.grpc:grpc-inprocess
  dependency-version: 1.74.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: io.grpc:grpc-util
  dependency-version: 1.74.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>

.github/ISSUE_TEMPLATE/bug_report.md

@@ -15,8 +15,8 @@ assignees: ''
 A clear and concise description of what the bug is.
 
 **To Reproduce**
-Steps or reproducer to reproduce the behavior.
+Steps or reproducer to reproduce the behavior in a form of a unit test.
-This section *must* be provided, if not, the issue may not get attention.
+This section *must* be provided, if not, the issue may not get attention since the maintainers have very limited capacity.
 
 **Expected behavior**
 A clear and concise description of what you expected to happen.

.github/workflows/build-main.yml

@@ -42,8 +42,8 @@ jobs:
           - 17
           - 21
         etcd:
-          - gcr.io/etcd-development/etcd:v3.5.10
+          - quay.io/coreos/etcd:v3.5.21
-          - gcr.io/etcd-development/etcd:v3.4.27
+          - quay.io/coreos/etcd:v3.6.0
     uses: ./.github/workflows/build.yml
     with:
       javaVersion: "${{ matrix.java-version }}"

.github/workflows/build-pr.yml

@@ -43,8 +43,8 @@ jobs:
           - 17
           - 21
         etcd:
-          - gcr.io/etcd-development/etcd:v3.5.10
+          - quay.io/coreos/etcd:v3.5.21
-          - gcr.io/etcd-development/etcd:v3.4.27
+          - quay.io/coreos/etcd:v3.6.0
     uses: ./.github/workflows/build.yml
     with:
       javaVersion: "${{ matrix.java-version }}"

.github/workflows/checks.yml

@@ -34,4 +34,4 @@ jobs:
       - name: 'Checkout'
         uses: actions/checkout@v4
       - name: 'Dependency Review'
-        uses: actions/dependency-review-action@v3
+        uses: actions/dependency-review-action@v4

.github/workflows/stale.yml

@@ -29,10 +29,11 @@ jobs:
         repo-token: ${{ secrets.GITHUB_TOKEN }}
         days-before-stale: 60
         days-before-close: 7
+        only-labels: 'waiting-for-feedbacks'
         stale-issue-label: stale
-        exempt-issue-label: never-stale
+        exempt-issue-labels: never-stale
         stale-pr-label: stale
-        exempt-pr-label: never-stale
+        exempt-pr-labels: never-stale
         stale-issue-message: |
           This issue is stale because it has been open 60 days with no activity.
           Remove stale label or comment or this will be closed in 7 days.

OWNERS

@@ -3,5 +3,4 @@
 approvers:
   - fanminshi       # Fanmin Shi <fanmin.shi@coreos.com>
   - lburgazzoli     # Luca Burgazzoli <lburgazzoli@gmail.com>
-  - xiang90         # Xiang Li <xiang.li@coreos.com>
+  - vorburger       # Michael Vorburger <mike@vorburger.ch>
-  - heyitsanthony   # Anthony Romano <anthony.romano@coreos.com>

README.md

@@ -10,7 +10,7 @@ jetcd is the official java client for [etcd](https://github.com/etcd-io/etcd) v3
 
 ## Java Versions
 
-Java 8 or above is required.
+Java 11 or above is required.
 
 ## Download
 
@@ -111,7 +111,7 @@ The project is tested against a three node `etcd` setup started with the Launche
 
 ```sh
 $ ./gradlew test
-````
+```
 
 ### Troubleshooting
 
@@ -128,3 +128,5 @@ See [CONTRIBUTING](https://github.com/etcd-io/jetcd/blob/master/CONTRIBUTING.md)
 ## License
 
 jetcd is under the Apache 2.0 license. See the [LICENSE](https://github.com/etcd-io/jetcd/blob/master/LICENSE) file for details.
+
+

build.gradle

@@ -63,7 +63,7 @@ subprojects {
     }
 
     compileJava {
-        options.release = 8
+        options.release = 11
     }
 
     test {
@@ -76,6 +76,13 @@ subprojects {
             maxFailures = 5
         }
     }
+    testing {
+        suites {
+            test {
+                useJUnitJupiter()
+            }
+        }
+    }
 
     testlogger {
         theme 'mocha-parallel'
@@ -83,6 +90,11 @@ subprojects {
     }
 
     tasks.register('allDeps', DependencyReportTask)
+
+    tasks.withType(AbstractArchiveTask).configureEach {
+        preserveFileTimestamps = false
+        reproducibleFileOrder = true
+    }
 }
 
 

gradle/libs.versions.toml

@@ -1,32 +1,34 @@
 [versions]
-grpc = "1.60.0"
+grpc = "1.74.0"
-log4j = "2.22.0"
+log4j = "2.25.1"
-mockito = "5.8.0"
+mockito = "5.18.0"
-slf4j = "2.0.9"
+slf4j = "2.0.17"
-guava = "33.0.0-jre"
+guava = "33.4.8-jre"
-assertj = "3.24.2"
+assertj = "3.27.3"
-junit = "5.10.1"
+junit = "5.13.4"
-testcontainers = "1.19.3"
+testcontainers = "1.21.3"
 protoc = "3.25.1"
 failsafe = "3.3.2"
-awaitility = "4.2.0"
+awaitility = "4.3.0"
-commonsIo = "2.15.1"
+commonsIo = "2.20.0"
-commonCompress = "1.25.0"
+commonCompress = "1.28.0"
 autoService = "1.1.1"
-errorprone = "2.23.0"
+errorprone = "2.30.0"
-vertx = "4.5.1"
+vertx = "5.0.1"
-picocli = "4.7.5"
+picocli = "4.7.7"
-restAssured = "5.4.0"
+restAssured = "5.5.5"
+javaxAnnotation = "1.3.2"
 
-versionsPlugin = "0.50.0"
+versionsPlugin = "0.52.0"
-errorPronePlugin = "3.1.0"
+errorPronePlugin = "4.0.1"
-spotlessPlugin = "6.23.3"
+spotlessPlugin = "6.25.0"
 shadowPlugin = "8.1.1"
 testLoggerPlugin = "4.0.0"
-protobufPlugin = "0.9.4"
+protobufPlugin = "0.9.5"
-nexusPublishPlugin = "1.3.0"
+nexusPublishPlugin = "2.0.0"
-axionReleasePlugin = "1.16.1"
+axionReleasePlugin = "1.18.17"
-testRetryPlugin = "1.5.8"
+testRetryPlugin = "1.6.2"
+
 
 [libraries]
 slf4j = { module = "org.slf4j:slf4j-api", version.ref = "slf4j" }
@@ -64,6 +66,7 @@ log4j12 = { module = "org.apache.logging.log4j:log4j-1.2-api", version.ref = "lo
 autoServiceAnnotations = { module = "com.google.auto.service:auto-service-annotations", version.ref = "autoService"}
 autoServiceProcessor = { module = "com.google.auto.service:auto-service", version.ref = "autoService"}
 
+javaxAnnotation = { module = "javax.annotation:javax.annotation-api", version.ref = "javaxAnnotation" }
 
 errorprone = { module = "com.google.errorprone:error_prone_core", version.ref = "errorprone" }
 errorproneAnnotations = { module = "com.google.errorprone:error_prone_annotations", version.ref = "errorprone" }
@@ -75,6 +78,7 @@ grpcTest = [ "grpcInprocess"]
 log4j    = [ "log4jApi", "log4jCore", "log4jSlf4j", "log4j12" ]
 mockito  = [ "mockitoCore", "mockitoJunit" ]
 testing  = ["junit", "assertj", "mockitoCore", "mockitoJunit"]
+javax    = [ "javaxAnnotation" ]
 
 
 [plugins]

gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,6 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.5-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.14-bin.zip
 networkTimeout=10000
 validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME

gradlew

@@ -15,6 +15,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+# SPDX-License-Identifier: Apache-2.0
+#
 
 ##############################################################################
 #
@@ -55,7 +57,7 @@
 #       Darwin, MinGW, and NonStop.
 #
 #   (3) This script is generated from the Groovy template
-#       https://github.com/gradle/gradle/blob/HEAD/subprojects/plugins/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
+#       https://github.com/gradle/gradle/blob/HEAD/platforms/jvm/plugins-application/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
 #       within the Gradle project.
 #
 #       You can find Gradle at https://github.com/gradle/gradle/.
@@ -84,7 +86,7 @@ done
 # shellcheck disable=SC2034
 APP_BASE_NAME=${0##*/}
 # Discard cd standard output in case $CDPATH is set (https://github.com/gradle/gradle/issues/25036)
-APP_HOME=$( cd "${APP_HOME:-./}" > /dev/null && pwd -P ) || exit
+APP_HOME=$( cd -P "${APP_HOME:-./}" > /dev/null && printf '%s\n' "$PWD" ) || exit
 
 # Use the maximum available, or set MAX_FD != -1 to use that value.
 MAX_FD=maximum
@@ -112,7 +114,7 @@ case "$( uname )" in                #(
   NONSTOP* )        nonstop=true ;;
 esac
 
-CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
+CLASSPATH="\\\"\\\""
 
 
 # Determine the Java command to use to start the JVM.
@@ -203,7 +205,7 @@ fi
 DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
 
 # Collect all arguments for the java command:
-#   * DEFAULT_JVM_OPTS, JAVA_OPTS, JAVA_OPTS, and optsEnvironmentVar are not allowed to contain shell fragments,
+#   * DEFAULT_JVM_OPTS, JAVA_OPTS, and optsEnvironmentVar are not allowed to contain shell fragments,
 #     and any embedded shellness will be escaped.
 #   * For example: A user cannot expect ${Hostname} to be expanded, as it is an environment variable and will be
 #     treated as '${Hostname}' itself on the command line.
@@ -211,7 +213,7 @@ DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
 set -- \
         "-Dorg.gradle.appname=$APP_BASE_NAME" \
         -classpath "$CLASSPATH" \
-        org.gradle.wrapper.GradleWrapperMain \
+        -jar "$APP_HOME/gradle/wrapper/gradle-wrapper.jar" \
         "$@"
 
 # Stop when "xargs" is not available.

gradlew.bat

@@ -1,92 +1,94 @@
-@rem
+@rem
-@rem Copyright 2015 the original author or authors.
+@rem Copyright 2015 the original author or authors.
-@rem
+@rem
-@rem Licensed under the Apache License, Version 2.0 (the "License");
+@rem Licensed under the Apache License, Version 2.0 (the "License");
-@rem you may not use this file except in compliance with the License.
+@rem you may not use this file except in compliance with the License.
-@rem You may obtain a copy of the License at
+@rem You may obtain a copy of the License at
-@rem
+@rem
-@rem      https://www.apache.org/licenses/LICENSE-2.0
+@rem      https://www.apache.org/licenses/LICENSE-2.0
-@rem
+@rem
-@rem Unless required by applicable law or agreed to in writing, software
+@rem Unless required by applicable law or agreed to in writing, software
-@rem distributed under the License is distributed on an "AS IS" BASIS,
+@rem distributed under the License is distributed on an "AS IS" BASIS,
-@rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+@rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-@rem See the License for the specific language governing permissions and
+@rem See the License for the specific language governing permissions and
-@rem limitations under the License.
+@rem limitations under the License.
-@rem
+@rem
-
+@rem SPDX-License-Identifier: Apache-2.0
-@if "%DEBUG%"=="" @echo off
+@rem
-@rem ##########################################################################
+
-@rem
+@if "%DEBUG%"=="" @echo off
-@rem  Gradle startup script for Windows
+@rem ##########################################################################
-@rem
+@rem
-@rem ##########################################################################
+@rem  Gradle startup script for Windows
-
+@rem
-@rem Set local scope for the variables with windows NT shell
+@rem ##########################################################################
-if "%OS%"=="Windows_NT" setlocal
+
-
+@rem Set local scope for the variables with windows NT shell
-set DIRNAME=%~dp0
+if "%OS%"=="Windows_NT" setlocal
-if "%DIRNAME%"=="" set DIRNAME=.
+
-@rem This is normally unused
+set DIRNAME=%~dp0
-set APP_BASE_NAME=%~n0
+if "%DIRNAME%"=="" set DIRNAME=.
-set APP_HOME=%DIRNAME%
+@rem This is normally unused
-
+set APP_BASE_NAME=%~n0
-@rem Resolve any "." and ".." in APP_HOME to make it shorter.
+set APP_HOME=%DIRNAME%
-for %%i in ("%APP_HOME%") do set APP_HOME=%%~fi
+
-
+@rem Resolve any "." and ".." in APP_HOME to make it shorter.
-@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+for %%i in ("%APP_HOME%") do set APP_HOME=%%~fi
-set DEFAULT_JVM_OPTS="-Xmx64m" "-Xms64m"
+
-
+@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
-@rem Find java.exe
+set DEFAULT_JVM_OPTS="-Xmx64m" "-Xms64m"
-if defined JAVA_HOME goto findJavaFromJavaHome
+
-
+@rem Find java.exe
-set JAVA_EXE=java.exe
+if defined JAVA_HOME goto findJavaFromJavaHome
-%JAVA_EXE% -version >NUL 2>&1
+
-if %ERRORLEVEL% equ 0 goto execute
+set JAVA_EXE=java.exe
-
+%JAVA_EXE% -version >NUL 2>&1
-echo.
+if %ERRORLEVEL% equ 0 goto execute
-echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+
-echo.
+echo. 1>&2
-echo Please set the JAVA_HOME variable in your environment to match the
+echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. 1>&2
-echo location of your Java installation.
+echo. 1>&2
-
+echo Please set the JAVA_HOME variable in your environment to match the 1>&2
-goto fail
+echo location of your Java installation. 1>&2
-
+
-:findJavaFromJavaHome
+goto fail
-set JAVA_HOME=%JAVA_HOME:"=%
+
-set JAVA_EXE=%JAVA_HOME%/bin/java.exe
+:findJavaFromJavaHome
-
+set JAVA_HOME=%JAVA_HOME:"=%
-if exist "%JAVA_EXE%" goto execute
+set JAVA_EXE=%JAVA_HOME%/bin/java.exe
-
+
-echo.
+if exist "%JAVA_EXE%" goto execute
-echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME%
+
-echo.
+echo. 1>&2
-echo Please set the JAVA_HOME variable in your environment to match the
+echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME% 1>&2
-echo location of your Java installation.
+echo. 1>&2
-
+echo Please set the JAVA_HOME variable in your environment to match the 1>&2
-goto fail
+echo location of your Java installation. 1>&2
-
+
-:execute
+goto fail
-@rem Setup the command line
+
-
+:execute
-set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar
+@rem Setup the command line
-
+
-
+set CLASSPATH=
-@rem Execute Gradle
+
-"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %*
+
-
+@rem Execute Gradle
-:end
+"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" -jar "%APP_HOME%\gradle\wrapper\gradle-wrapper.jar" %*
-@rem End local scope for the variables with windows NT shell
+
-if %ERRORLEVEL% equ 0 goto mainEnd
+:end
-
+@rem End local scope for the variables with windows NT shell
-:fail
+if %ERRORLEVEL% equ 0 goto mainEnd
-rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of
+
-rem the _cmd.exe /c_ return code!
+:fail
-set EXIT_CODE=%ERRORLEVEL%
+rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of
-if %EXIT_CODE% equ 0 set EXIT_CODE=1
+rem the _cmd.exe /c_ return code!
-if not ""=="%GRADLE_EXIT_CONSOLE%" exit %EXIT_CODE%
+set EXIT_CODE=%ERRORLEVEL%
-exit /b %EXIT_CODE%
+if %EXIT_CODE% equ 0 set EXIT_CODE=1
-
+if not ""=="%GRADLE_EXIT_CONSOLE%" exit %EXIT_CODE%
-:mainEnd
+exit /b %EXIT_CODE%
-if "%OS%"=="Windows_NT" endlocal
+
-
+:mainEnd
-:omega
+if "%OS%"=="Windows_NT" endlocal
+
+:omega

jetcd-core/src/main/java/io/etcd/jetcd/ClientBuilder.java

@@ -37,6 +37,7 @@ import io.grpc.Metadata;
 import io.grpc.netty.GrpcSslContexts;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
+import io.vertx.core.Vertx;
 
 import com.google.common.base.Strings;
 
@@ -68,6 +69,7 @@ public final class ClientBuilder implements Cloneable {
     private Duration retryMaxDuration;
     private Duration connectTimeout;
     private boolean waitForReady = true;
+    private Vertx vertx;
 
     ClientBuilder() {
     }
@@ -707,6 +709,30 @@ public final class ClientBuilder implements Cloneable {
         return this;
     }
 
+    /**
+     * Gets the Vertx instance.
+     *
+     * @return the vertx instance.
+     */
+    public Vertx vertx() {
+        return vertx;
+    }
+
+    /**
+     * configure Vertx instance.
+     *
+     * @param  vertx                    Vertx instance to use.
+     * @return                          this builder to train
+     * @throws IllegalArgumentException if vertx is null
+     */
+    public ClientBuilder vertx(Vertx vertx) {
+        Preconditions.checkArgument(vertx != null, "vertx can't be null");
+
+        this.vertx = vertx;
+
+        return this;
+    }
+
     /**
      * build a new Client.
      *

jetcd-core/src/main/java/io/etcd/jetcd/KV.java

@@ -26,6 +26,7 @@ import io.etcd.jetcd.options.CompactOption;
 import io.etcd.jetcd.options.DeleteOption;
 import io.etcd.jetcd.options.GetOption;
 import io.etcd.jetcd.options.PutOption;
+import io.etcd.jetcd.options.TxnOption;
 import io.etcd.jetcd.support.CloseableClient;
 
 /**
@@ -115,4 +116,12 @@ public interface KV extends CloseableClient {
      * @return a Txn
      */
     Txn txn();
+
+    /**
+     * creates a transaction.
+     *
+     * @param  option TxnOption
+     * @return        a Txn
+     */
+    Txn txn(TxnOption option);
 }

jetcd-core/src/main/java/io/etcd/jetcd/Watch.java

@@ -226,6 +226,11 @@ public interface Watch extends CloseableClient {
         @Override
         void close();
 
+        /**
+         * Returns if watcher is already closed
+         */
+        boolean isClosed();
+
         /**
          * Requests the latest revision processed and propagates it to listeners
          */

jetcd-core/src/main/java/io/etcd/jetcd/impl/ClientConnectionManager.java

@@ -61,6 +61,10 @@ final class ClientConnectionManager {
         } else {
             this.executorService = builder.executorService();
         }
+
+        if (builder.vertx() != null) {
+            this.vertx = builder.vertx();
+        }
     }
 
     ManagedChannel getChannel() {
@@ -187,7 +191,7 @@ final class ClientConnectionManager {
         if (builder.loadBalancerPolicy() != null) {
             channelBuilder.defaultLoadBalancingPolicy(builder.loadBalancerPolicy());
         } else {
-            channelBuilder.defaultLoadBalancingPolicy("pick_first");
+            channelBuilder.defaultLoadBalancingPolicy("round_robin");
         }
 
         if (builder.headers() != null) {

jetcd-core/src/main/java/io/etcd/jetcd/impl/ElectionImpl.java

@@ -32,6 +32,7 @@ import io.etcd.jetcd.election.NoLeaderException;
 import io.etcd.jetcd.election.NotLeaderException;
 import io.etcd.jetcd.election.ProclaimResponse;
 import io.etcd.jetcd.election.ResignResponse;
+import io.etcd.jetcd.support.Errors;
 import io.etcd.jetcd.support.Util;
 import io.grpc.StatusRuntimeException;
 
@@ -51,6 +52,27 @@ final class ElectionImpl extends Impl implements Election {
         this.namespace = connectionManager.getNamespace();
     }
 
+    // Election operations are done in a context where a client is trying to implement
+    // some fault tolerance related use case; in that type of context, it makes sense to always
+    // apply require leader, since we don't want a client connected to a non-raft-leader server
+    // have an election method just go silent if the server the client happens to be connected to
+    // becomes partitioned from the actual raft-leader server in the etcd servers cluster:
+    // in that scenario and without required leader, an attempt to campaign could block forever
+    // not because some other client is already an election leader, but because the server the client
+    // is connected to is partitioned and can't tell.
+    // With require leader, in that case the call will fail and we give
+    // the client the ability to (a) know (b) retry on a different server.
+    // The retry on a different server should happen automatically if the connection manager is using
+    // a round robin strategy.
+    //
+    // Beware in the context of this election API, the word "leader" is overloaded.
+    // In the paragraph above when we say "raft-leader" we are talking about the etcd server that is a leader
+    // of the etcd servers cluster according to raft, we are not talking about the client that
+    // happens to be the leader of an election using the election API in this file.
+    private VertxElectionGrpc.ElectionVertxStub stubWithLeader() {
+        return Util.applyRequireLeader(true, stub);
+    }
+
     @Override
     public CompletableFuture<CampaignResponse> campaign(ByteSequence electionName, long leaseId, ByteSequence proposal) {
         requireNonNull(electionName, "election name should not be null");
@@ -62,10 +84,11 @@ final class ElectionImpl extends Impl implements Election {
             .setLease(leaseId)
             .build();
 
-        return completable(
+        return wrapConvertException(
-            stub.campaign(request),
+            execute(
-            CampaignResponse::new,
+                () -> stubWithLeader().campaign(request),
-            this::convertException);
+                CampaignResponse::new,
+                Errors::isRetryableForNoSafeRedoOp));
     }
 
     @Override
@@ -84,10 +107,11 @@ final class ElectionImpl extends Impl implements Election {
             .setValue(ByteString.copyFrom(proposal.getBytes()))
             .build();
 
-        return completable(
+        return wrapConvertException(
-            stub.proclaim(request),
+            execute(
-            ProclaimResponse::new,
+                () -> stubWithLeader().proclaim(request),
-            this::convertException);
+                ProclaimResponse::new,
+                Errors::isRetryableForNoSafeRedoOp));
     }
 
     @Override
@@ -98,10 +122,11 @@ final class ElectionImpl extends Impl implements Election {
             .setName(Util.prefixNamespace(electionName, namespace))
             .build();
 
-        return completable(
+        return wrapConvertException(
-            stub.leader(request),
+            execute(
-            r -> new LeaderResponse(r, namespace),
+                () -> stubWithLeader().leader(request),
-            this::convertException);
+                response -> new LeaderResponse(response, namespace),
+                Errors::isRetryableForNoSafeRedoOp));
     }
 
     @Override
@@ -110,13 +135,13 @@ final class ElectionImpl extends Impl implements Election {
         requireNonNull(listener, "listener should not be null");
 
         LeaderRequest request = LeaderRequest.newBuilder()
-            .setName(ByteString.copyFrom(electionName.getBytes()))
+            .setName(Util.prefixNamespace(electionName, namespace))
             .build();
 
-        stub.observe(request)
+        stubWithLeader().observeWithHandler(request,
-            .handler(value -> listener.onNext(new LeaderResponse(value, namespace)))
+            value -> listener.onNext(new LeaderResponse(value, namespace)),
-            .endHandler(ignored -> listener.onCompleted())
+            ignored -> listener.onCompleted(),
-            .exceptionHandler(error -> listener.onError(toEtcdException(error)));
+            error -> listener.onError(toEtcdException(error)));
     }
 
     @Override
@@ -133,25 +158,42 @@ final class ElectionImpl extends Impl implements Election {
                     .build())
             .build();
 
-        return completable(
+        return wrapConvertException(
-            stub.resign(request),
+            execute(
-            ResignResponse::new,
+                () -> stubWithLeader().resign(request),
-            this::convertException);
+                ResignResponse::new,
+                Errors::isRetryableForNoSafeRedoOp));
     }
 
-    private Throwable convertException(Throwable e) {
+    private <S> CompletableFuture<S> wrapConvertException(CompletableFuture<S> future) {
-        if (e instanceof StatusRuntimeException) {
+        return future.exceptionally(e -> {
-            StatusRuntimeException exception = (StatusRuntimeException) e;
+            throw convertException(e);
-            String description = exception.getStatus().getDescription();
+        });
-            // different APIs use different messages. we cannot distinguish missing leader error otherwise,
+    }
-            // because communicated status is always UNKNOWN
-            if ("election: not leader".equals(description)) {
-                return new NotLeaderException();
-            } else if ("election: no leader".equals(description)) {
-                return new NoLeaderException();
-            }
-        }
 
+    private RuntimeException convertException(Throwable e) {
+        Throwable cause = e;
+        while (cause != null) {
+            if (cause instanceof StatusRuntimeException) {
+                StatusRuntimeException exception = (StatusRuntimeException) cause;
+                String description = exception.getStatus().getDescription();
+                // different APIs use different messages. we cannot distinguish missing leader error otherwise,
+                // because communicated status is always UNKNOWN
+                if ("election: not leader".equals(description)) {
+                    // Candidate is not a leader at the moment.
+                    // Note there is a one letter difference, but this exception type is not the same as
+                    // NoLeaderException.
+                    return new NotLeaderException();
+                }
+                if ("election: no leader".equals(description)) {
+                    // Leader for given election does not exist.
+                    // Note there is a one letter difference, but this exception type is not the same as
+                    // NotLeaderException.
+                    return new NoLeaderException();
+                }
+            }
+            cause = cause.getCause();
+        }
         return toEtcdException(e);
     }
 }

jetcd-core/src/main/java/io/etcd/jetcd/impl/Impl.java

@@ -89,9 +89,11 @@ abstract class Impl {
      */
     protected <S, T> CompletableFuture<T> execute(
         Supplier<Future<S>> supplier,
-        Function<S, T> resultConvert) {
+        Function<S, T> resultConvert,
+        boolean autoRetry) {
 
-        return execute(supplier, resultConvert, Errors::isRetryable);
+        return execute(supplier, resultConvert,
+            autoRetry ? Errors::isRetryableForSafeRedoOp : Errors::isRetryableForNoSafeRedoOp);
     }
 
     /**

jetcd-core/src/main/java/io/etcd/jetcd/impl/KVImpl.java

@@ -33,6 +33,7 @@ import io.etcd.jetcd.options.CompactOption;
 import io.etcd.jetcd.options.DeleteOption;
 import io.etcd.jetcd.options.GetOption;
 import io.etcd.jetcd.options.PutOption;
+import io.etcd.jetcd.options.TxnOption;
 import io.etcd.jetcd.support.Errors;
 import io.etcd.jetcd.support.Requests;
 
@@ -65,7 +66,7 @@ final class KVImpl extends Impl implements KV {
         return execute(
             () -> stub.put(Requests.mapPutRequest(key, value, option, namespace)),
             response -> new PutResponse(response, namespace),
-            Errors::isRetryable);
+            option.isAutoRetry() ? Errors::isRetryableForSafeRedoOp : Errors::isRetryableForNoSafeRedoOp);
     }
 
     @Override
@@ -81,7 +82,7 @@ final class KVImpl extends Impl implements KV {
         return execute(
             () -> stub.range(Requests.mapRangeRequest(key, option, namespace)),
             response -> new GetResponse(response, namespace),
-            Errors::isRetryable);
+            Errors::isRetryableForSafeRedoOp);
     }
 
     @Override
@@ -97,7 +98,7 @@ final class KVImpl extends Impl implements KV {
         return execute(
             () -> stub.deleteRange(Requests.mapDeleteRequest(key, option, namespace)),
             response -> new DeleteResponse(response, namespace),
-            Errors::isRetryable);
+            option.isAutoRetry() ? Errors::isRetryableForSafeRedoOp : Errors::isRetryableForNoSafeRedoOp);
     }
 
     @Override
@@ -116,15 +117,21 @@ final class KVImpl extends Impl implements KV {
         return execute(
             () -> stub.compact(request),
             CompactResponse::new,
-            Errors::isRetryable);
+            Errors::isRetryableForSafeRedoOp);
     }
 
     @Override
     public Txn txn() {
+        return txn(TxnOption.DEFAULT);
+    }
+
+    @Override
+    public Txn txn(TxnOption option) {
         return TxnImpl.newTxn(
             request -> execute(
                 () -> stub.txn(request),
-                response -> new TxnResponse(response, namespace), Errors::isRetryable),
+                response -> new TxnResponse(response, namespace),
+                option.isAutoRetry() ? Errors::isRetryableForSafeRedoOp : Errors::isRetryableForNoSafeRedoOp),
             namespace);
     }
 }

jetcd-core/src/main/java/io/etcd/jetcd/impl/LeaseImpl.java

@@ -84,7 +84,8 @@ final class LeaseImpl extends Impl implements Lease {
                 LeaseGrantRequest.newBuilder()
                     .setTTL(ttl)
                     .build()),
-            LeaseGrantResponse::new);
+            LeaseGrantResponse::new,
+            true);
     }
 
     @Override
@@ -94,7 +95,8 @@ final class LeaseImpl extends Impl implements Lease {
                 LeaseGrantRequest.newBuilder()
                     .setTTL(ttl)
                     .build()),
-            LeaseGrantResponse::new);
+            LeaseGrantResponse::new,
+            true);
     }
 
     @Override
@@ -104,7 +106,8 @@ final class LeaseImpl extends Impl implements Lease {
                 LeaseRevokeRequest.newBuilder()
                     .setID(leaseId)
                     .build()),
-            LeaseRevokeResponse::new);
+            LeaseRevokeResponse::new,
+            true);
     }
 
     @Override
@@ -118,7 +121,8 @@ final class LeaseImpl extends Impl implements Lease {
 
         return execute(
             () -> this.stub.leaseTimeToLive(leaseTimeToLiveRequest),
-            LeaseTimeToLiveResponse::new);
+            LeaseTimeToLiveResponse::new,
+            true);
     }
 
     @Override
@@ -148,19 +152,21 @@ final class LeaseImpl extends Impl implements Lease {
         final LeaseKeepAliveRequest req = LeaseKeepAliveRequest.newBuilder().setID(leaseId).build();
 
         leaseStub
-            .leaseKeepAlive(s -> {
+            .leaseKeepAliveWithHandler(
-                ref.set(s);
+                s -> {
-                s.write(req);
+                    ref.set(s);
-            })
+                    s.write(req);
-            .handler(r -> {
+                },
-                if (r.getTTL() != 0) {
+                r -> {
-                    future.complete(new LeaseKeepAliveResponse(r));
+                    if (r.getTTL() != 0) {
-                } else {
+                        future.complete(new LeaseKeepAliveResponse(r));
-                    future.completeExceptionally(
+                    } else {
-                        newEtcdException(ErrorCode.NOT_FOUND, "etcdserver: requested lease not found"));
+                        future.completeExceptionally(
-                }
+                            newEtcdException(ErrorCode.NOT_FOUND, "etcdserver: requested lease not found"));
-            })
+                    }
-            .exceptionHandler(future::completeExceptionally);
+                },
+                null,
+                future::completeExceptionally);
 
         return future.whenComplete((r, t) -> ref.get().end(req));
     }
@@ -194,9 +200,11 @@ final class LeaseImpl extends Impl implements Lease {
 
         @Override
         public void doStart() {
-            leaseStub.leaseKeepAlive(this::writeHandler)
+            leaseStub.leaseKeepAliveWithHandler(
-                .handler(this::handleResponse)
+                this::writeHandler,
-                .exceptionHandler(this::handleException);
+                this::handleResponse,
+                null,
+                this::handleException);
         }
 
         @Override
@@ -262,11 +270,13 @@ final class LeaseImpl extends Impl implements Lease {
             }
         }
 
-        private synchronized void handleException(Throwable unused) {
+        private synchronized void handleException(Throwable throwable) {
             if (!this.isRunning()) {
                 return;
             }
 
+            keepAlives.values().forEach(ka -> ka.onError(throwable));
+
             restart = connectionManager().vertx().setTimer(
                 500,
                 l -> {

jetcd-core/src/main/java/io/etcd/jetcd/impl/LockImpl.java

@@ -34,6 +34,22 @@ final class LockImpl extends Impl implements Lock {
     private final VertxLockGrpc.LockVertxStub stub;
     private final ByteSequence namespace;
 
+    // Lock operations are done in a context where a client is trying to implement
+    // some strict mutual exclusion use case; in that type of context, it makes sense to always
+    // apply require leader, since we don't want a client connected to a non-raft-leader server
+    // to have a lock method just go silent if the server the client happens to be connected to
+    // becomes partitioned from the actual raft-leader server in the etcd servers cluster:
+    // in that scenario and without required leader, an attempt to lock could block forever
+    // not because some other client is already holding a lock, but because the server the client
+    // is connected to is partitioned and can't tell.
+    // With require leader, in that case the call will fail and the client has the ability to
+    // (a) know (b) retry on a different server.
+    // The retry on a different server should happen automatically if the connection manager is using
+    // a round robin strategy.
+    private VertxLockGrpc.LockVertxStub stubWithLeader() {
+        return Util.applyRequireLeader(true, stub);
+    }
+
     LockImpl(ClientConnectionManager connectionManager) {
         super(connectionManager);
 
@@ -51,9 +67,9 @@ final class LockImpl extends Impl implements Lock {
             .build();
 
         return execute(
-            () -> stub.lock(request),
+            () -> stubWithLeader().lock(request),
             response -> new LockResponse(response, namespace),
-            Errors::isRetryable);
+            Errors::isRetryableForSafeRedoOp);
     }
 
     @Override
@@ -65,8 +81,8 @@ final class LockImpl extends Impl implements Lock {
             .build();
 
         return execute(
-            () -> stub.unlock(request),
+            () -> stubWithLeader().unlock(request),
             UnlockResponse::new,
-            Errors::isRetryable);
+            Errors::isRetryableForSafeRedoOp);
     }
 }

jetcd-core/src/main/java/io/etcd/jetcd/impl/MaintenanceImpl.java

@@ -119,19 +119,20 @@ final class MaintenanceImpl extends Impl implements Maintenance {
         final CompletableFuture<Long> answer = new CompletableFuture<>();
         final AtomicLong bytes = new AtomicLong(0);
 
-        this.stub.snapshot(SnapshotRequest.getDefaultInstance())
+        this.stub.snapshotWithHandler(
-            .handler(r -> {
+            SnapshotRequest.getDefaultInstance(),
+            r -> {
                 try {
                     r.getBlob().writeTo(outputStream);
                     bytes.addAndGet(r.getBlob().size());
                 } catch (IOException e) {
                     answer.completeExceptionally(toEtcdException(e));
                 }
-            })
+            },
-            .endHandler(event -> {
+            event -> {
                 answer.complete(bytes.get());
-            })
+            },
-            .exceptionHandler(e -> {
+            e -> {
                 answer.completeExceptionally(toEtcdException(e));
             });
 
@@ -141,9 +142,10 @@ final class MaintenanceImpl extends Impl implements Maintenance {
     @Override
     public void snapshot(StreamObserver<io.etcd.jetcd.maintenance.SnapshotResponse> observer) {
 
-        this.stub.snapshot(SnapshotRequest.getDefaultInstance())
+        this.stub.snapshotWithHandler(
-            .handler(r -> observer.onNext(new io.etcd.jetcd.maintenance.SnapshotResponse(r)))
+            SnapshotRequest.getDefaultInstance(),
-            .endHandler(event -> observer.onCompleted())
+            r -> observer.onNext(new io.etcd.jetcd.maintenance.SnapshotResponse(r)),
-            .exceptionHandler(e -> observer.onError(toEtcdException(e)));
+            event -> observer.onCompleted(),
+            e -> observer.onError(toEtcdException(e)));
     }
 }

jetcd-core/src/main/java/io/etcd/jetcd/impl/WatchImpl.java

@@ -23,13 +23,17 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
-import org.checkerframework.checker.nullness.qual.NonNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.etcd.jetcd.ByteSequence;
 import io.etcd.jetcd.Watch;
-import io.etcd.jetcd.api.*;
+import io.etcd.jetcd.api.VertxWatchGrpc;
+import io.etcd.jetcd.api.WatchCancelRequest;
+import io.etcd.jetcd.api.WatchCreateRequest;
+import io.etcd.jetcd.api.WatchProgressRequest;
+import io.etcd.jetcd.api.WatchRequest;
+import io.etcd.jetcd.api.WatchResponse;
 import io.etcd.jetcd.common.exception.ErrorCode;
 import io.etcd.jetcd.common.exception.EtcdException;
 import io.etcd.jetcd.options.OptionsUtil;
@@ -37,7 +41,6 @@ import io.etcd.jetcd.options.WatchOption;
 import io.etcd.jetcd.support.Errors;
 import io.etcd.jetcd.support.Util;
 import io.grpc.Status;
-import io.vertx.core.streams.ReadStream;
 import io.vertx.core.streams.WriteStream;
 
 import com.google.common.base.Strings;
@@ -46,7 +49,10 @@ import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListeningScheduledExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 
-import static io.etcd.jetcd.common.exception.EtcdExceptionFactory.*;
+import static io.etcd.jetcd.common.exception.EtcdExceptionFactory.newClosedWatchClientException;
+import static io.etcd.jetcd.common.exception.EtcdExceptionFactory.newCompactedException;
+import static io.etcd.jetcd.common.exception.EtcdExceptionFactory.newEtcdException;
+import static io.etcd.jetcd.common.exception.EtcdExceptionFactory.toEtcdException;
 
 /**
  * watch Implementation.
@@ -117,10 +123,8 @@ final class WatchImpl extends Impl implements Watch {
         private final Listener listener;
         private final AtomicBoolean closed;
 
-        //private StreamObserver<WatchRequest> stream;
         private final AtomicReference<WriteStream<WatchRequest>> wstream;
         private final AtomicBoolean started;
-        private ReadStream<WatchResponse> rstream;
         private long revision;
         private long id;
 
@@ -132,7 +136,6 @@ final class WatchImpl extends Impl implements Watch {
 
             this.started = new AtomicBoolean();
             this.wstream = new AtomicReference<>();
-            this.rstream = null;
             this.id = -1;
             this.revision = this.option.getRevision();
         }
@@ -143,7 +146,8 @@ final class WatchImpl extends Impl implements Watch {
         //
         // ************************
 
-        boolean isClosed() {
+        @Override
+        public boolean isClosed() {
             return this.closed.get() || WatchImpl.this.closed.get();
         }
 
@@ -153,7 +157,7 @@ final class WatchImpl extends Impl implements Watch {
             }
 
             if (started.compareAndSet(false, true)) {
-                // id is not really useful today but it may be in etcd 3.4
+                // id is not really useful today, but it may be in etcd 3.4
                 id = -1;
 
                 WatchCreateRequest.Builder builder = WatchCreateRequest.newBuilder()
@@ -165,7 +169,7 @@ final class WatchImpl extends Impl implements Watch {
                     .map(endKey -> Util.prefixNamespaceToRangeEnd(endKey, namespace))
                     .ifPresent(builder::setRangeEnd);
 
-                if (!option.getEndKey().isPresent() && option.isPrefix()) {
+                if (option.getEndKey().isEmpty() && option.isPrefix()) {
                     ByteSequence endKey = OptionsUtil.prefixEndOf(key);
                     builder.setRangeEnd(Util.prefixNamespaceToRangeEnd(endKey, namespace));
                 }
@@ -178,13 +182,15 @@ final class WatchImpl extends Impl implements Watch {
                     builder.addFilters(WatchCreateRequest.FilterType.NOPUT);
                 }
 
-                rstream = Util.applyRequireLeader(option.withRequireLeader(), stub).watchWithExceptionHandler(stream -> {
+                var ignored = Util.applyRequireLeader(option.withRequireLeader(), stub)
-                    wstream.set(stream);
+                    .watchWithHandler(
-                    stream.write(WatchRequest.newBuilder().setCreateRequest(builder).build());
+                        stream -> {
-                }, this::onError);
+                            wstream.set(stream);
-
+                            stream.write(WatchRequest.newBuilder().setCreateRequest(builder).build());
-                rstream.handler(this::onNext);
+                        },
-                rstream.endHandler(event -> onCompleted());
+                        this::onNext,
+                        event -> onCompleted(),
+                        this::onError);
             }
         }
 
@@ -350,7 +356,7 @@ final class WatchImpl extends Impl implements Watch {
         private void reschedule() {
             Futures.addCallback(executor.schedule(this::resume, 500, TimeUnit.MILLISECONDS), new FutureCallback<Object>() {
                 @Override
-                public void onFailure(@NonNull Throwable t) {
+                public void onFailure(Throwable t) {
                     LOG.warn("scheduled resume failed", t);
                 }
 

jetcd-core/src/main/java/io/etcd/jetcd/kv/GetResponse.java

@@ -58,7 +58,11 @@ public class GetResponse extends AbstractResponse<RangeResponse> {
     }
 
     /**
-     * Returns the number of keys within the range when requested.
+     * Returns the number of keys within the range requested.
+     * Note this value is never affected by filtering options (limit, min or max created or modified revisions)
+     * Count is the count for keys on the range part of a request.
+     * Filters for limit and created or modified revision ranges restrict the
+     * returned KVs, but not the count.
      *
      * @return count.
      */

jetcd-core/src/main/java/io/etcd/jetcd/options/DeleteOption.java

@@ -29,11 +29,13 @@ public final class DeleteOption {
     private final ByteSequence endKey;
     private final boolean prevKV;
     private final boolean prefix;
+    private final boolean autoRetry;
 
-    private DeleteOption(ByteSequence endKey, boolean prevKV, boolean prefix) {
+    private DeleteOption(ByteSequence endKey, boolean prevKV, boolean prefix, final boolean autoRetry) {
         this.endKey = endKey;
         this.prevKV = prevKV;
         this.prefix = prefix;
+        this.autoRetry = autoRetry;
     }
 
     public Optional<ByteSequence> getEndKey() {
@@ -49,10 +51,25 @@ public final class DeleteOption {
         return prevKV;
     }
 
+    /**
+     * Whether to treat this deletion as deletion by prefix
+     *
+     * @return true if deletion by prefix.
+     */
     public boolean isPrefix() {
         return prefix;
     }
 
+    /**
+     * Whether to treat a delete operation as idempotent from the point of view of automated retries.
+     * Note under failure scenarios this may mean a single delete is attempted more than once.
+     *
+     * @return true if automated retries should happen.
+     */
+    public boolean isAutoRetry() {
+        return autoRetry;
+    }
+
     /**
      * Returns the builder.
      *
@@ -65,6 +82,11 @@ public final class DeleteOption {
         return builder();
     }
 
+    /**
+     * Returns the builder.
+     *
+     * @return the builder
+     */
     public static Builder builder() {
         return new Builder();
     }
@@ -73,6 +95,7 @@ public final class DeleteOption {
         private ByteSequence endKey;
         private boolean prevKV = false;
         private boolean prefix = false;
+        private boolean autoRetry = false;
 
         private Builder() {
         }
@@ -144,8 +167,22 @@ public final class DeleteOption {
             return this;
         }
 
+        /**
+         * When autoRetry is set, the delete operation is treated as idempotent from the point of view of automated retries.
+         * Note under some failure scenarios true may make a delete operation be attempted more than once, where
+         * a first attempt succeeded but its result did not reach the client; by default (autoRetry=false),
+         * the client won't retry since it is not safe to assume on such a failure the operation did not happen.
+         * Requesting withAutoRetry means the client is explicitly asking for retry nevertheless.
+         *
+         * @return builder
+         */
+        public Builder withAutoRetry() {
+            this.autoRetry = true;
+            return this;
+        }
+
         public DeleteOption build() {
-            return new DeleteOption(endKey, prevKV, prefix);
+            return new DeleteOption(endKey, prevKV, prefix, autoRetry);
         }
 
     }

jetcd-core/src/main/java/io/etcd/jetcd/options/GetOption.java

@@ -20,6 +20,7 @@ import java.util.Optional;
 
 import io.etcd.jetcd.ByteSequence;
 import io.etcd.jetcd.KV;
+import io.etcd.jetcd.kv.GetResponse;
 
 import static java.util.Objects.requireNonNull;
 
@@ -76,56 +77,158 @@ public final class GetOption {
     /**
      * Get the maximum number of keys to return for a get request.
      *
+     * <p>
+     * Note this filter does not affect the count field in GetResponse.
+     * {@link GetResponse#getCount()} always counts the number of keys matched on a range, independent of filters.
+     * </p>
+     *
+     *
      * @return the maximum number of keys to return.
      */
     public long getLimit() {
         return this.limit;
     }
 
+    /**
+     * Get the end key for a range request
+     *
+     * @return the end key for a range request
+     */
     public Optional<ByteSequence> getEndKey() {
         return Optional.ofNullable(this.endKey);
     }
 
+    /**
+     * Get the revision for the request
+     *
+     * @return the revision for the request
+     */
     public long getRevision() {
         return revision;
     }
 
+    /**
+     * Get the sort order for the request
+     *
+     * @return the sort order for the request
+     */
     public SortOrder getSortOrder() {
         return sortOrder;
     }
 
+    /**
+     * Get the sort field for the request
+     *
+     * @return the sort field for the request
+     */
     public SortTarget getSortField() {
         return sortTarget;
     }
 
+    /**
+     * Return if the consistency level for this request is "serializable".
+     * Note serializable is a lower than default consistency, and implies
+     * the possibility of getting stale data.
+     *
+     * @return true if this request is only serializable consistency
+     */
     public boolean isSerializable() {
         return serializable;
     }
 
+    /**
+     * True if this request should get only keys in a range and there is no
+     * need to retrieve the values.
+     *
+     * @return true if only get keys
+     */
     public boolean isKeysOnly() {
         return keysOnly;
     }
 
+    /**
+     * True if this request should only populate the count of keys matching
+     * a range, and no other data.
+     *
+     * @return true if only get the count of keys
+     */
     public boolean isCountOnly() {
         return countOnly;
     }
 
+    /**
+     * Only populate results for keys that match a
+     * minimum value for a created revision.
+     *
+     * <p>
+     * Note this filter does not affect the count field in GetResponse.
+     * {@link GetResponse#getCount()} always counts the number of keys matched on a range, independent of filters.
+     * For the same reason, it would be meaningless to mix setting a min create revision option
+     * with the count only option.
+     * </p>
+     *
+     * @return minimum created revision to match, or zero for any.
+     */
     public long getMinCreateRevision() {
         return this.minCreateRevision;
     }
 
+    /**
+     * Only populate results for keys that match a
+     * maximum value for a created revision.
+     *
+     * <p>
+     * Note this filter does not affect the count field in GetResponse.
+     * {@link GetResponse#getCount()} always counts the number of keys matched on a range, independent of filters.
+     * For the same reason, it would be meaningless to mix setting a max create revision option
+     * with the count only option.
+     * </p>
+     *
+     * @return maximum created revision to match, or zero for any.
+     */
     public long getMaxCreateRevision() {
         return this.maxCreateRevision;
     }
 
+    /**
+     * Only populate results for keys that match a
+     * minimum value for a modified revision.
+     *
+     * <p>
+     * Note this filter does not affect the count field in GetResponse.
+     * {@link GetResponse#getCount()} always counts the number of keys matched on a range, independent of filters.
+     * For the same reason, it would be meaningless to mix setting a min mod revision option
+     * with the count only option.
+     * </p>
+     *
+     * @return minimum modified revision to match, or zero for any.
+     */
     public long getMinModRevision() {
         return this.minModRevision;
     }
 
+    /**
+     * Only populate results for keys that match a
+     * maximum value for a modified revision.
+     *
+     * <p>
+     * Note this filter does not affect the count field in GetResponse.
+     * {@link GetResponse#getCount()} always counts the number of keys matched on a range, independent of filters.
+     * For the same reason, it would be meaningless to mix setting a max mod revision option
+     * with the count only option.
+     * </p>
+     *
+     * @return maximum modified revision to match, or zero for any.
+     */
     public long getMaxModRevision() {
         return this.maxModRevision;
     }
 
+    /**
+     * True if this Get request should do prefix match
+     *
+     * @return true if this Get request should do prefix match
+     */
     public boolean isPrefix() {
         return prefix;
     }
@@ -176,6 +279,13 @@ public final class GetOption {
         /**
          * Limit the number of keys to return for a get request. By default is 0 - no limitation.
          *
+         * <p>
+         * Note this filter does not affect the count field in GetResponse.
+         * {@link GetResponse#getCount()} always counts the number of keys matched on a range, independent of filters.
+         * For the same reason, it would be meaningless to mix setting this option
+         * with the count only option.
+         * </p>
+         *
          * @param  limit the maximum number of keys to return for a get request.
          * @return       builder
          */
@@ -229,7 +339,9 @@ public final class GetOption {
          * <p>
          * Get requests are linearizable by
          * default. For better performance, a serializable get request is served locally without needing
-         * to reach consensus with other nodes in the cluster.
+         * to reach consensus with other nodes in the cluster. Note this is a tradeoff with strict
+         * consistency so it should be used with care in situations where reading stale
+         * is acceptable.
          *
          * @param  serializable is the get request a serializable get request.
          * @return              builder
@@ -316,10 +428,17 @@ public final class GetOption {
         }
 
         /**
-         * Limit returned keys to those with create revision greater than the provided value.
+         * Limit returned keys to those with create revision greater or equal than the provided value.
          * min_create_revision is the lower bound for returned key create revisions; all keys with
          * lesser create revisions will be filtered away.
          *
+         * <p>
+         * Note this filter does not affect the count field in GetResponse.
+         * {@link GetResponse#getCount()} always counts the number of keys matched on a range, independent of filters.
+         * For the same reason, it would be meaningless to mix setting this option
+         * with the count only option.
+         * </p>
+         *
          * @param  createRevision create revision
          * @return                builder
          */
@@ -329,10 +448,17 @@ public final class GetOption {
         }
 
         /**
-         * Limit returned keys to those with create revision less than the provided value.
+         * Limit returned keys to those with create revision less or equal than the provided value.
          * max_create_revision is the upper bound for returned key create revisions; all keys with
          * greater create revisions will be filtered away.
          *
+         * <p>
+         * Note this filter does not affect the count field in GetResponse.
+         * {@link GetResponse#getCount()} always counts the number of keys matched on a range, independent of filters.
+         * For the same reason, it would be meaningless to mix setting this option
+         * with the count only option.
+         * </p>
+         *
          * @param  createRevision create revision
          * @return                builder
          */
@@ -342,10 +468,17 @@ public final class GetOption {
         }
 
         /**
-         * Limit returned keys to those with mod revision greater than the provided value.
+         * Limit returned keys to those with mod revision greater or equal than the provided value.
          * min_mod_revision is the lower bound for returned key mod revisions; all keys with lesser mod
          * revisions will be filtered away.
          *
+         * <p>
+         * Note this filter does not affect the count field in GetResponse.
+         * {@link GetResponse#getCount()} always counts the number of keys matched on a range, independent of filters.
+         * For the same reason, it would be meaningless to mix setting this option
+         * with the count only option.
+         * </p>
+         *
          * @param  modRevision mod revision
          * @return             this builder instance
          */
@@ -355,10 +488,17 @@ public final class GetOption {
         }
 
         /**
-         * Limit returned keys to those with mod revision less than the provided value. max_mod_revision
+         * Limit returned keys to those with mod revision less or equal than the provided value. max_mod_revision
          * is the upper bound for returned key mod revisions; all keys with greater mod revisions will
          * be filtered away.
          *
+         * <p>
+         * Note this filter does not affect the count field in GetResponse.
+         * {@link GetResponse#getCount()} always counts the number of keys matched on a range, independent of filters.
+         * For the same reason, it would be meaningless to mix setting this option
+         * with the count only option.
+         * </p>
+         *
          * @param  modRevision mod revision
          * @return             this builder instance
          */

jetcd-core/src/main/java/io/etcd/jetcd/options/PutOption.java

@@ -26,10 +26,12 @@ public final class PutOption {
 
     private final long leaseId;
     private final boolean prevKV;
+    private final boolean autoRetry;
 
-    private PutOption(long leaseId, boolean prevKV) {
+    private PutOption(long leaseId, boolean prevKV, boolean autoRetry) {
         this.leaseId = leaseId;
         this.prevKV = prevKV;
+        this.autoRetry = autoRetry;
     }
 
     /**
@@ -50,6 +52,16 @@ public final class PutOption {
         return this.prevKV;
     }
 
+    /**
+     * Whether to treat a put operation as idempotent from the point of view of automated retries.
+     * Note under failure scenarios this may mean a single put executes more than once.
+     *
+     * @return true if automated retries should happen.
+     */
+    public boolean isAutoRetry() {
+        return autoRetry;
+    }
+
     /**
      * Returns the builder.
      *
@@ -73,6 +85,7 @@ public final class PutOption {
 
         private long leaseId = 0L;
         private boolean prevKV = false;
+        private boolean autoRetry = false;
 
         private Builder() {
         }
@@ -100,13 +113,28 @@ public final class PutOption {
             return this;
         }
 
+        /**
+         * When autoRetry is set, treat this put as idempotent from the point of view of automated retries.
+         * Note under some failure scenarios autoRetry=true may make a put operation execute more than once, where
+         * a first attempt succeeded but its result did not reach the client; by default (autoRetry=false),
+         * the client won't retry since it is not safe to assume on such a failure that the operation did not happen
+         * in the server.
+         * Requesting withAutoRetry means the client is explicitly asking for retry nevertheless.
+         *
+         * @return builder
+         */
+        public Builder withAutoRetry() {
+            this.autoRetry = true;
+            return this;
+        }
+
         /**
          * build the put option.
          *
          * @return the put option
          */
         public PutOption build() {
-            return new PutOption(this.leaseId, this.prevKV);
+            return new PutOption(this.leaseId, this.prevKV, this.autoRetry);
         }
 
     }

jetcd-core/src/main/java/io/etcd/jetcd/options/TxnOption.java

@@ -0,0 +1,54 @@
+package io.etcd.jetcd.options;
+
+public final class TxnOption {
+    public static final TxnOption DEFAULT = builder().build();
+
+    private final boolean autoRetry;
+
+    private TxnOption(final boolean autoRetry) {
+        this.autoRetry = autoRetry;
+    }
+
+    /**
+     * Whether to treat a txn operation as idempotent from the point of view of automated retries.
+     *
+     * @return true if automated retries should happen.
+     */
+    public boolean isAutoRetry() {
+        return autoRetry;
+    }
+
+    /**
+     * Returns the builder.
+     *
+     * @return the builder
+     */
+    public static TxnOption.Builder builder() {
+        return new TxnOption.Builder();
+    }
+
+    public static final class Builder {
+        private boolean autoRetry = false;
+
+        private Builder() {
+        }
+
+        /**
+         * When autoRetry is set, the txn operation is treated as idempotent from the point of view of automated retries.
+         * Note under some failure scenarios true may make a txn operation be attempted and/or execute more than once, where
+         * a first attempt executed but its result status did not reach the client; by default (autoRetry=false),
+         * the client won't retry since it is not safe to assume on such a failure the operation did not happen.
+         * Requesting withAutoRetry means the client is explicitly asking for retry nevertheless.
+         *
+         * @return builder
+         */
+        public Builder withAutoRetry() {
+            this.autoRetry = true;
+            return this;
+        }
+
+        public TxnOption build() {
+            return new TxnOption(autoRetry);
+        }
+    }
+}

jetcd-core/src/main/java/io/etcd/jetcd/resolver/AbstractNameResolver.java

@@ -19,7 +19,6 @@ package io.etcd.jetcd.resolver;
 import java.net.URI;
 import java.util.List;
 import java.util.Objects;
-import java.util.concurrent.Executor;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -30,8 +29,6 @@ import io.grpc.Attributes;
 import io.grpc.EquivalentAddressGroup;
 import io.grpc.NameResolver;
 import io.grpc.Status;
-import io.grpc.internal.GrpcUtil;
-import io.grpc.internal.SharedResourceHolder;
 
 import com.google.common.base.Preconditions;
 
@@ -47,7 +44,6 @@ public abstract class AbstractNameResolver extends NameResolver {
     private volatile boolean shutdown;
     private volatile boolean resolving;
 
-    private Executor executor;
     private Listener listener;
 
     public AbstractNameResolver(URI targetUri) {
@@ -69,7 +65,6 @@ public abstract class AbstractNameResolver extends NameResolver {
     public void start(Listener listener) {
         synchronized (lock) {
             Preconditions.checkState(this.listener == null, "already started");
-            this.executor = SharedResourceHolder.get(GrpcUtil.SHARED_CHANNEL_EXECUTOR);
             this.listener = Objects.requireNonNull(listener, "listener");
             resolve();
         }
@@ -86,21 +81,13 @@ public abstract class AbstractNameResolver extends NameResolver {
             return;
         }
         shutdown = true;
-
-        synchronized (lock) {
-            if (executor != null) {
-                executor = SharedResourceHolder.release(GrpcUtil.SHARED_CHANNEL_EXECUTOR, executor);
-            }
-        }
     }
 
     private void resolve() {
         if (resolving || shutdown) {
             return;
         }
-        synchronized (lock) {
+        doResolve();
-            executor.execute(this::doResolve);
-        }
     }
 
     private void doResolve() {

jetcd-core/src/main/java/io/etcd/jetcd/support/Errors.java

@@ -26,8 +26,18 @@ public final class Errors {
     private Errors() {
     }
 
-    public static boolean isRetryable(Status status) {
+    // isRetryable implementation for idempotent operations.
-        return Status.UNAVAILABLE.getCode().equals(status.getCode()) || isInvalidTokenError(status);
+    public static boolean isRetryableForSafeRedoOp(Status status) {
+        return Status.UNAVAILABLE.getCode().equals(status.getCode()) || isAlwaysSafeToRetry(status);
+    }
+
+    // isRetryable implementation for non-idempotent operations
+    public static boolean isRetryableForNoSafeRedoOp(Status status) {
+        return isAlwaysSafeToRetry(status);
+    }
+
+    public static boolean isAlwaysSafeToRetry(Status status) {
+        return isInvalidTokenError(status) || isAuthStoreExpired(status);
     }
 
     public static boolean isInvalidTokenError(Throwable e) {

jetcd-core/src/test/java/io/etcd/jetcd/impl/ClientBuilderTest.java

@@ -50,6 +50,11 @@ public class ClientBuilderTest {
         assertThatThrownBy(() -> Client.builder().endpoints((URI) null)).isInstanceOf(NullPointerException.class);
     }
 
+    @Test
+    public void testVertx_Null() {
+        assertThatThrownBy(() -> Client.builder().vertx(null)).isInstanceOf(IllegalArgumentException.class);
+    }
+
     @Test
     public void testEndPoints_Verify_Empty() {
         assertThatThrownBy(() -> Client.builder().endpoints(new URI(""))).isInstanceOf(IllegalArgumentException.class);

jetcd-core/src/test/java/io/etcd/jetcd/impl/ElectionTest.java

@@ -197,6 +197,39 @@ public class ElectionTest {
         leaseClient.revoke(leaseId).get();
     }
 
+    @Test
+    public void testObserveExistingLeader() throws Exception {
+        final AtomicInteger electionsSeen = new AtomicInteger(0);
+        ByteSequence electionName = ByteSequence.from(randomString(), StandardCharsets.UTF_8);
+
+        long leaseId = leaseClient.grant(10).get().getID();
+
+        ByteSequence proposal = ByteSequence.from(randomString(), StandardCharsets.UTF_8);
+        electionClient.campaign(electionName, leaseId, proposal)
+            .get(OPERATION_TIMEOUT, TimeUnit.SECONDS);
+
+        electionClient.observe(electionName, new Election.Listener() {
+            @Override
+            public void onNext(LeaderResponse response) {
+                electionsSeen.incrementAndGet();
+            }
+
+            @Override
+            public void onError(Throwable error) {
+            }
+
+            @Override
+            public void onCompleted() {
+            }
+        });
+
+        TestUtil.waitForCondition(
+            () -> electionsSeen.get() == 1, OPERATION_TIMEOUT * 1000,
+            "Observer did not receive expected notifications, got: " + electionsSeen.get());
+
+        leaseClient.revoke(leaseId).get();
+    }
+
     @Test
     public void testSynchronizationBarrier() throws Exception {
         final int threadCount = 5;

jetcd-core/src/test/java/io/etcd/jetcd/impl/KVTest.java

@@ -316,7 +316,7 @@ public class KVTest {
                 for (int i = 0; i < putCount; ++i) {
                     ByteSequence value = ByteSequence
                         .from(Integer.toString(i), StandardCharsets.UTF_8);
-                    customClient.getKVClient().put(key, value).join();
+                    customClient.getKVClient().put(key, value, PutOption.builder().withAutoRetry().build()).join();
                 }
             });
 

jetcd-core/src/test/java/io/etcd/jetcd/impl/LeaseUnitTest.java

@@ -48,6 +48,7 @@ import io.grpc.stub.StreamObserver;
 import static io.etcd.jetcd.common.exception.EtcdExceptionFactory.toEtcdException;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.Mockito.after;
 import static org.mockito.Mockito.timeout;
@@ -57,6 +58,7 @@ import static org.mockito.Mockito.verify;
 @Timeout(value = 1, unit = TimeUnit.MINUTES)
 @ExtendWith(MockitoExtension.class)
 public class LeaseUnitTest {
+    private static final long TIME_OUT_SECONDS = 30;
 
     private Lease leaseCli;
     private AtomicReference<StreamObserver<LeaseKeepAliveResponse>> responseObserverRef;
@@ -230,8 +232,10 @@ public class LeaseUnitTest {
 
     @Test
     public void testKeepAliveResetOnStreamErrors() {
-        final StreamObserver<io.etcd.jetcd.lease.LeaseKeepAliveResponse> observer = Observers.observer(response -> {
+        final AtomicReference<Throwable> errorRecorder = new AtomicReference<>();
-        });
+        final StreamObserver<io.etcd.jetcd.lease.LeaseKeepAliveResponse> observer = Observers.<io.etcd.jetcd.lease.LeaseKeepAliveResponse> builder()
+            .onError(errorRecorder::set)
+            .build();
 
         try (CloseableClient client = this.leaseCli.keepAlive(LEASE_ID_1, observer)) {
             Throwable t = Status.ABORTED.asRuntimeException();
@@ -241,6 +245,7 @@ public class LeaseUnitTest {
 
             // expect keep alive requests are still sending even with reset.
             verify(this.requestStreamObserverMock, timeout(2000).atLeast(3)).onNext(argThat(hasLeaseID(LEASE_ID_1)));
+            await().atMost(TIME_OUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> assertThat(errorRecorder.get()).isNotNull());
         }
     }
 

jetcd-core/src/test/java/io/etcd/jetcd/impl/UtilTest.java

@@ -39,16 +39,18 @@ class UtilTest {
     }
 
     @Test
-    public void testAuthErrorIsNotRetryable() {
+    public void testAuthErrorIsRetryable() {
         Status authErrorStatus = Status.UNAUTHENTICATED
             .withDescription("etcdserver: invalid auth token");
         Status status = Status.fromThrowable(new StatusException(authErrorStatus));
-        assertThat(Errors.isRetryable(status)).isTrue();
+        assertThat(Errors.isRetryableForNoSafeRedoOp(status)).isTrue();
+        assertThat(Errors.isRetryableForSafeRedoOp(status)).isTrue();
     }
 
     @Test
     public void testUnavailableErrorIsRetryable() {
         Status status = Status.fromThrowable(new StatusException(Status.UNAVAILABLE));
-        assertThat(Errors.isRetryable(status)).isTrue();
+        assertThat(Errors.isRetryableForNoSafeRedoOp(status)).isFalse();
+        assertThat(Errors.isRetryableForSafeRedoOp(status)).isTrue();
     }
 }

jetcd-core/src/test/java/io/etcd/jetcd/impl/WatchUnitTest.java

@@ -427,8 +427,7 @@ public class WatchUnitTest {
             assertThat(ref.get()).isNotNull();
             assertThat(ref.get()).isInstanceOf(EtcdException.class)
                 .hasMessageContaining(Errors.NO_LEADER_ERROR_MESSAGE);
-            final WatchImpl.WatcherImpl wimpl = (WatchImpl.WatcherImpl) watcher;
+            assertThat(watcher.isClosed()).isTrue();
-            assertThat(wimpl.isClosed()).isTrue();
         }
     }
 }

jetcd-core/src/test/java/io/etcd/jetcd/impl/auth/AuthTokenRefreshTest.java

@@ -0,0 +1,97 @@
+package io.etcd.jetcd.impl.auth;
+
+import java.io.File;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KV;
+import io.etcd.jetcd.auth.Permission;
+import io.etcd.jetcd.impl.TestUtil;
+import io.etcd.jetcd.kv.GetResponse;
+import io.etcd.jetcd.test.EtcdClusterExtension;
+
+@Timeout(value = 30)
+public class AuthTokenRefreshTest {
+
+    @RegisterExtension
+    public static final EtcdClusterExtension cluster = EtcdClusterExtension.builder()
+        .withNodes(1)
+        .withSsl(true)
+        .withAdditionalArgs(
+            List.of(
+                "--auth-token",
+                "jwt,pub-key=/etc/ssl/etcd/server.pem,priv-key=/etc/ssl/etcd/server-key.pem,sign-method=RS256,ttl=300s"))
+        .build();
+
+    private static final ByteSequence key = TestUtil.bytesOf("key");
+    private static final ByteSequence keyEnd = TestUtil.bytesOf("key1");
+    private static final ByteSequence user = TestUtil.bytesOf("root");
+    private static final ByteSequence password = TestUtil.randomByteSequence();
+
+    @Test
+    public void testTokenIsRefreshedWhenRevisionsIsOld() throws Exception {
+        setUpEnvironment();
+
+        Client authClient = createAuthClient();
+        KV kvClient = authClient.getKVClient();
+
+        ByteSequence value = TestUtil.randomByteSequence();
+        kvClient.put(key, value).get(1, TimeUnit.SECONDS);
+
+        //read the created key to generate a new jwt token
+        GetResponse getResponse = kvClient.get(key).get(1, TimeUnit.SECONDS);
+        Assertions.assertEquals(1, getResponse.getKvs().size());
+        Assertions.assertEquals(key, getResponse.getKvs().get(0).getKey());
+        Assertions.assertEquals(value, getResponse.getKvs().get(0).getValue());
+
+        //update auth database revision by adding a new role thus invalidating existing token of the kv client
+        authClient.getAuthClient().roleAdd(TestUtil.bytesOf("newRole")).get(1, TimeUnit.SECONDS);
+
+        //kv client should automatically refresh the old token and read the key
+        getResponse = kvClient.get(key).get(1, TimeUnit.SECONDS);
+
+        Assertions.assertEquals(1, getResponse.getKvs().size());
+        Assertions.assertEquals(key, getResponse.getKvs().get(0).getKey());
+        Assertions.assertEquals(value, getResponse.getKvs().get(0).getValue());
+    }
+
+    private void setUpEnvironment() throws Exception {
+        final File caFile = new File(Objects.requireNonNull(getClass().getResource("/ssl/cert/ca.pem")).toURI());
+
+        Client client = TestUtil.client(cluster)
+            .authority("etcd0")
+            .sslContext(b -> b.trustManager(caFile))
+            .build();
+
+        // enable authentication to enforce usage of access token
+        ByteSequence role = TestUtil.bytesOf("root");
+        client.getAuthClient().roleAdd(role).get();
+        client.getAuthClient().userAdd(user, password).get();
+        // grant access only to given key
+        client.getAuthClient().roleGrantPermission(role, key, keyEnd, Permission.Type.READWRITE).get();
+        client.getAuthClient().userGrantRole(user, role).get();
+        client.getAuthClient().authEnable().get();
+
+        client.close();
+    }
+
+    private Client createAuthClient() throws Exception {
+        final File caFile = new File(Objects.requireNonNull(getClass().getResource("/ssl/cert/ca.pem")).toURI());
+
+        return TestUtil.client(cluster)
+            .user(user)
+            .password(password)
+            .authority("etcd0")
+            .sslContext(b -> b.trustManager(caFile))
+            .build();
+    }
+
+}

jetcd-grpc/build.gradle

@@ -22,6 +22,7 @@ dependencies {
     api libs.slf4j
     api libs.vertxGrpc
     api libs.bundles.grpc
+    api libs.bundles.javax
 }
 
 protobuf {

jetcd-launcher/src/main/java/io/etcd/jetcd/launcher/Etcd.java

@@ -28,7 +28,7 @@ import org.testcontainers.containers.Network;
 import com.google.common.base.Strings;
 
 public final class Etcd {
-    public static final String CONTAINER_IMAGE = "gcr.io/etcd-development/etcd:v3.5.10";
+    public static final String CONTAINER_IMAGE = "quay.io/coreos/etcd:v3.5.14";
     public static final int ETCD_CLIENT_PORT = 2379;
     public static final int ETCD_PEER_PORT = 2380;
     public static final String ETCD_DATA_DIR = "/data.etcd";
@@ -58,6 +58,7 @@ public final class Etcd {
         private List<String> additionalArgs;
         private Network network;
         private boolean shouldMountDataDirectory = true;
+        private String user;
 
         public Builder withClusterName(String clusterName) {
             this.clusterName = clusterName;
@@ -114,12 +115,18 @@ public final class Etcd {
                 debug,
                 additionalArgs,
                 network != null ? network : Network.SHARED,
-                shouldMountDataDirectory);
+                shouldMountDataDirectory,
+                user);
         }
 
         public Builder withMountedDataDirectory(boolean shouldMountDataDirectory) {
             this.shouldMountDataDirectory = shouldMountDataDirectory;
             return this;
         }
+
+        public Builder withUser(String user) {
+            this.user = user;
+            return this;
+        }
     }
 }

jetcd-launcher/src/main/java/io/etcd/jetcd/launcher/EtcdClusterImpl.java

@@ -43,7 +43,8 @@ public class EtcdClusterImpl implements EtcdCluster {
         boolean debug,
         Collection<String> additionalArgs,
         Network network,
-        boolean shouldMountDataDirectory) {
+        boolean shouldMountDataDirectory,
+        String user) {
 
         this.clusterName = clusterName;
         this.endpoints = IntStream.range(0, nodes)
@@ -57,7 +58,8 @@ public class EtcdClusterImpl implements EtcdCluster {
                 .withDebug(debug)
                 .withAdditionalArgs(additionalArgs)
                 .withNetwork(network)
-                .withShouldMountDataDirectory(shouldMountDataDirectory))
+                .withShouldMountDataDirectory(shouldMountDataDirectory)
+                .withUser(user))
             .collect(toList());
     }
 

jetcd-launcher/src/main/java/io/etcd/jetcd/launcher/EtcdContainer.java

@@ -63,6 +63,7 @@ public class EtcdContainer extends GenericContainer<EtcdContainer> {
     private Path dataDirectory;
     private Collection<String> additionalArgs;
     private boolean shouldMountDataDirectory = true;
+    private String user;
 
     public EtcdContainer(String image, String node, Collection<String> nodes) {
         super(image);
@@ -102,6 +103,18 @@ public class EtcdContainer extends GenericContainer<EtcdContainer> {
         return self();
     }
 
+    /**
+     * Optional values are {@code [ user | user:group | uid | uid:gid | user:gid | uid:group ]}.
+     * See <a href="https://docs.docker.com/engine/reference/run/#user">User</a> .
+     *
+     * @param  user Refer to {@link com.github.dockerjava.api.command.CreateContainerCmd#withUser(String)}
+     * @return      self container.
+     */
+    public EtcdContainer withUser(String user) {
+        this.user = user;
+        return self();
+    }
+
     @Override
     protected void configure() {
         if (!configured.compareAndSet(false, true)) {
@@ -120,9 +133,13 @@ public class EtcdContainer extends GenericContainer<EtcdContainer> {
         withEnv("ETCD_LOG_LEVEL", this.debug ? "debug" : "info");
         withEnv("ETCD_LOGGER", "zap");
 
-        String user = System.getenv("TC_USER");
+        String tempUser = this.user;
-        if (user != null) {
+        if (tempUser == null) {
-            withCreateContainerCmdModifier(c -> c.withUser(user));
+            tempUser = System.getenv("TC_USER");
+        }
+        if (tempUser != null) {
+            String finalUser = tempUser;
+            withCreateContainerCmdModifier(c -> c.withUser(finalUser));
         }
 
         if (ssl) {

jetcd-launcher/src/test/resources/log4j2.xml

@@ -27,9 +27,6 @@
   <Loggers>
     <!-- package loggers -->
     <Logger name="org.testcontainers" level="INFO"/>
-    <Logger name="🐳 [gcr.io/etcd-development/etcd:v3.3]" level="WARN"/>
-    <Logger name="🐳 [gcr.io/etcd-development/etcd:v3.4]" level="WARN"/>
-    <Logger name="🐳 [gcr.io/etcd-development/etcd:v3.4.7]" level="WARN"/>
     <Logger name="io.etcd.jetcd.internal.infrastructure" level="INFO"/>
     <Logger name="io.etcd.jetcd.launcher.EtcdCluster" level="WARN"/>
 

jetcd-test/src/main/java/io/etcd/jetcd/test/EtcdClusterExtension.java

@@ -168,6 +168,11 @@ public class EtcdClusterExtension implements BeforeAllCallback, BeforeEachCallba
             return this;
         }
 
+        public Builder withUser(String user) {
+            builder.withUser(user);
+            return this;
+        }
+
         public EtcdClusterExtension build() {
             return new EtcdClusterExtension(builder.build());
         }

jetcd-test/src/main/java/io/etcd/jetcd/test/EtcdClusterNameResolver.java

@@ -20,7 +20,6 @@ import java.net.URI;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
-import java.util.concurrent.Executor;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -31,8 +30,6 @@ import io.grpc.Attributes;
 import io.grpc.EquivalentAddressGroup;
 import io.grpc.NameResolver;
 import io.grpc.Status;
-import io.grpc.internal.GrpcUtil;
-import io.grpc.internal.SharedResourceHolder;
 
 import com.google.common.base.Preconditions;
 
@@ -47,7 +44,6 @@ public class EtcdClusterNameResolver extends NameResolver {
     private volatile boolean shutdown;
     private volatile boolean resolving;
 
-    private Executor executor;
     private Listener listener;
 
     public EtcdClusterNameResolver(URI targetUri) {
@@ -65,7 +61,6 @@ public class EtcdClusterNameResolver extends NameResolver {
     public void start(Listener listener) {
         synchronized (lock) {
             Preconditions.checkState(this.listener == null, "already started");
-            this.executor = SharedResourceHolder.get(GrpcUtil.SHARED_CHANNEL_EXECUTOR);
             this.listener = Objects.requireNonNull(listener, "listener");
             resolve();
         }
@@ -82,21 +77,14 @@ public class EtcdClusterNameResolver extends NameResolver {
             return;
         }
         shutdown = true;
-
-        synchronized (lock) {
-            if (executor != null) {
-                executor = SharedResourceHolder.release(GrpcUtil.SHARED_CHANNEL_EXECUTOR, executor);
-            }
-        }
     }
 
     private void resolve() {
         if (resolving || shutdown) {
             return;
         }
-        synchronized (lock) {
+
-            executor.execute(this::doResolve);
+        doResolve();
-        }
     }
 
     private void doResolve() {