diff --git a/docs/modules/ROOT/pages/http-client.adoc b/docs/modules/ROOT/pages/http-client.adoc
index 9eb4125778..44a176cd22 100644
--- a/docs/modules/ROOT/pages/http-client.adoc
+++ b/docs/modules/ROOT/pages/http-client.adoc
@@ -742,3 +742,144 @@ To customize the default settings, you can configure `HttpClient` as follows:
 include::{examples-dir}/resolver/Application.java[lines=18..39]
 ----
 <1> The timeout of each DNS query performed by this resolver will be 500ms.
+
+[[http-client-spnego]]
+== SPNEGO Authentication
+Reactor Netty HttpClient supports SPNEGO (Kerberos) authentication, which is widely used in enterprise environments.
+SPNEGO (Simple and Protected GSSAPI Negotiation Mechanism) provides secure authentication over HTTP using Kerberos.
+
+==== How It Works
+SPNEGO authentication follows this HTTP authentication flow:
+
+. The client sends an HTTP request to a protected resource.
+. The server responds with `401 Unauthorized` and a `WWW-Authenticate: Negotiate` header.
+. The client generates a SPNEGO token based on its Kerberos ticket, and resends the request with an `Authorization: Negotiate <base64-encoded-token>` header.
+. The server validates the token and, if authentication is successful, returns 200 OK.
+
+If further negotiation is required, the server may return another 401 with additional data in the `WWW-Authenticate` header.
+
+==== JAAS-based Authenticator
+{examples-link}/spnego/jaas/Application.java
+----
+include::{examples-dir}/spnego/jaas/Application.java[lines=18..45]
+----
+<1> Configures the `jaas.conf`. A JAAS configuration file in Java for integrating with authentication backends such as Kerberos.
+<2> Configures the `krb5.conf`. krb5.conf is a Kerberos client configuration file used to define how the client locates and communicates with the Kerberos Key Distribution Center (KDC) for authentication.
+<3> Configures the SPNEGO jaas.conf. A JVM system property that enables detailed debug logging for Kerberos operations in Java.
+<4> `JaasAuthenticator` performs Kerberos authentication using a JAAS configuration (jaas.conf).
+<5> `SpnegoAuthProvider.Builder` supports the following configuration methods. Please refer to <<spnegoauthprovider-config>>.
+<6> `SpnegoAuthProvider`  generates a SPNEGO token from the Kerberos ticket. It automatically adds the `Authorization: Negotiate ...` header to HTTP requests. If the server responds with `401 Unauthorized` and includes `WWW-Authenticate: Negotiate`, the client will automatically reauthenticate and retry the request once.
+
+===== Example JAAS Configuration
+Specify the path to your JAAS configuration file using the `java.security.auth.login.config` system property.
+
+.`jaas.conf`
+[jaas,conf]
+----
+KerberosLogin {
+    com.sun.security.auth.module.Krb5LoginModule required
+    client=true
+    useKeyTab=true
+    keyTab="/path/to/test.keytab"
+    principal="test@EXAMPLE.COM"
+    doNotPrompt=true
+    debug=true;
+};
+----
+
+===== Example Kerberos Configuration
+Specify Kerberos realm and KDC information using the `java.security.krb5.conf` system property.
+
+.`krb5.conf`
+[krb5,conf]
+----
+[libdefaults]
+    default_realm = EXAMPLE.COM
+[realms]
+    EXAMPLE.COM = {
+        kdc = kdc.example.com
+    }
+[domain_realms]
+    .example.com = EXAMPLE.COM
+    example.com = EXAMPLE.COM
+----
+
+===== Configuration Example
+[jvm option]
+----
+-Djava.security.auth.login.config=/path/to/login.conf
+-Djava.security.krb5.conf=/path/to/krb5.conf
+----
+
+==== GSSCredential-based Authenticator
+For scenarios where you already have a `GSSCredential` available or want to avoid JAAS configuration, you can use `GssCredentialAuthenticator`:
+
+{examples-link}/spnego/gsscredential/Application.java
+----
+include::{examples-dir}/spnego/gsscredential/Application.java[lines=18..46]
+----
+<1> Obtain the `GSSCredential` through other means.
+<2> Configure the GSSCredential-based authenticator for SPNEGO authentication.
+
+This approach is useful when:
+- You want to reuse existing credentials
+- You need more control over credential management
+- JAAS configuration is not available or preferred
+
+==== Custom Authenticator Implementation
+For advanced scenarios where the provided authenticators don't meet your specific requirements, you can implement the `SpnegoAuthenticator` interface directly:
+
+----
+import org.ietf.jgss.GSSContext;
+import reactor.netty.http.client.SpnegoAuthenticator;
+import reactor.netty.http.client.SpnegoAuthProvider;
+
+public class CustomSpnegoAuthenticator implements SpnegoAuthenticator {
+
+    @Override
+    public GSSContext createContext(String serviceName, String remoteHost) throws Exception {
+        // Your custom authentication logic here
+        // This method should return a configured GSSContext
+        // for the specified service and remote host
+        // serviceName: e.g., "HTTP", "LDAP"
+        // remoteHost: target server hostname
+
+        return null; // Replace with actual GSSContext creation logic
+    }
+}
+
+// Usage with advanced configuration
+HttpClient client = HttpClient.create()
+	.spnego(
+		SpnegoAuthProvider.builder(new CustomSpnegoAuthenticator())
+			.serviceName("HTTP")  // Custom service name
+			.unauthorizedStatusCode(401)  // Custom status code
+			.resolveCanonicalHostname(true)  // Use canonical hostname
+			.build()
+	);
+----
+
+This approach is useful when you need:
+- Custom credential acquisition logic
+- Integration with third-party authentication systems
+- Special handling for token caching or refresh
+- Environment-specific authentication flows
+
+[[spnegoauthprovider-config]]
+==== SpnegoAuthProvider Configuration Options
+The `SpnegoAuthProvider.Builder` supports the following configuration Options:
+
+[width="100%",options="header"]
+|=======
+| Method | Default | Description | Example
+| `serviceName(String)` | "HTTP" | Service name for constructing service principal names (serviceName/hostname) | "HTTP", "LDAP"
+| `unauthorizedStatusCode(int)` | 401 | HTTP status code that triggers authentication retry | 401, 407
+| `resolveCanonicalHostname(boolean)` | false | Whether to use canonical hostname resolution via reverse DNS lookup | true for FQDN requirements
+|=======
+
+==== Notes
+- SPNEGO authentication is fully supported on Java 1.6 and above.
+- If authentication fails, check the server logs and client exception messages, and verify your Kerberos environment settings (realm, KDC, ticket, etc.).
+- `JaasAuthenticator` performs authentication through JAAS login configuration.
+- `GssCredentialAuthenticator` uses pre-existing `GSSCredential` objects, bypassing JAAS configuration.
+- For custom scenarios, implement the `SpnegoAuthenticator` interface to provide your own authentication logic.
diff --git a/reactor-netty-examples/src/main/java/reactor/netty/examples/documentation/http/client/spnego/gsscredential/Application.java b/reactor-netty-examples/src/main/java/reactor/netty/examples/documentation/http/client/spnego/gsscredential/Application.java
new file mode 100644
index 0000000000..e67ab5af09
--- /dev/null
+++ b/reactor-netty-examples/src/main/java/reactor/netty/examples/documentation/http/client/spnego/gsscredential/Application.java
@@ -0,0 +1,46 @@
+/*
+ * Copyright (c) 2025 VMware, Inc. or its affiliates, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package reactor.netty.examples.documentation.http.client.spnego.gsscredential;
+
+import org.ietf.jgss.GSSCredential;
+import reactor.netty.http.client.GssCredentialAuthenticator;
+import reactor.netty.http.client.HttpClient;
+import reactor.netty.http.client.SpnegoAuthProvider;
+
+public class Application {
+
+	public static void main(String[] args) {
+		// Assuming you have obtained a GSSCredential from elsewhere
+		GSSCredential credential = obtainGSSCredential(); // <1>
+
+		HttpClient client = HttpClient.create()
+				.spnego(
+						SpnegoAuthProvider.builder(new GssCredentialAuthenticator(credential)) // <2>
+								.build()
+				);
+
+		client.get()
+				.uri("http://protected.example.com/")
+				.responseSingle((res, content) -> content.asString())
+				.block();
+	}
+
+	private static GSSCredential obtainGSSCredential() {
+		// Implement your logic to obtain a GSSCredential
+		// This could involve using a Kerberos library or other means
+		return null;
+	}
+}
diff --git a/reactor-netty-examples/src/main/java/reactor/netty/examples/documentation/http/client/spnego/jaas/Application.java b/reactor-netty-examples/src/main/java/reactor/netty/examples/documentation/http/client/spnego/jaas/Application.java
new file mode 100644
index 0000000000..d011faa1d7
--- /dev/null
+++ b/reactor-netty-examples/src/main/java/reactor/netty/examples/documentation/http/client/spnego/jaas/Application.java
@@ -0,0 +1,46 @@
+/*
+ * Copyright (c) 2025 VMware, Inc. or its affiliates, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package reactor.netty.examples.documentation.http.client.spnego.jaas;
+
+import reactor.netty.http.client.HttpClient;
+import reactor.netty.http.client.JaasAuthenticator;
+import reactor.netty.http.client.SpnegoAuthProvider;
+import reactor.netty.http.client.SpnegoAuthenticator;
+
+public class Application {
+
+	public static void main(String[] args) {
+		System.setProperty("java.security.auth.login.config", "/path/to/jaas.conf"); // <1>
+		System.setProperty("java.security.krb5.conf", "/path/to/krb5.conf"); // <2>
+		System.setProperty("sun.security.krb5.debug", "true"); // <3>
+
+		SpnegoAuthenticator authenticator = new JaasAuthenticator("KerberosLogin"); // <4>
+		SpnegoAuthProvider provider = SpnegoAuthProvider.builder(authenticator)
+				.serviceName("HTTP")
+				.unauthorizedStatusCode(401)
+				.resolveCanonicalHostname(false)
+				.build();
+		HttpClient client = HttpClient.create()
+				.spnego(
+						provider // <5>
+				); // <6>
+
+		client.get()
+				.uri("http://protected.example.com/")
+				.responseSingle((res, content) -> content.asString())
+				.block();
+	}
+}
diff --git a/reactor-netty-http/src/main/java/reactor/netty/http/client/GssCredentialAuthenticator.java b/reactor-netty-http/src/main/java/reactor/netty/http/client/GssCredentialAuthenticator.java
new file mode 100644
index 0000000000..721ad6b5fd
--- /dev/null
+++ b/reactor-netty-http/src/main/java/reactor/netty/http/client/GssCredentialAuthenticator.java
@@ -0,0 +1,88 @@
+/*
+ * Copyright (c) 2025 VMware, Inc. or its affiliates, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package reactor.netty.http.client;
+
+import java.util.Objects;
+import org.ietf.jgss.GSSContext;
+import org.ietf.jgss.GSSCredential;
+import org.ietf.jgss.GSSManager;
+import org.ietf.jgss.GSSName;
+import org.ietf.jgss.Oid;
+
+/**
+ * A GSSCredential-based Authenticator implementation for use with SPNEGO providers.
+ * <p>
+ * This authenticator uses a pre-existing GSSCredential to create a GSSContext,
+ * bypassing the need for JAAS login configuration. This is useful when you already
+ * have obtained Kerberos credentials through other means or want more direct control
+ * over the authentication process.
+ * </p>
+ * <p>
+ * The GSSCredential should contain valid Kerberos credentials that can be used
+ * for SPNEGO authentication. The credential's lifetime and validity are managed
+ * externally to this authenticator.
+ * </p>
+ *
+ * <p>Example usage:</p>
+ * <pre>
+ *     GSSCredential credential = // ... obtain credential
+ *     GssCredentialAuthenticator authenticator = new GssCredentialAuthenticator(credential);
+ *     SpnegoAuthProvider provider = SpnegoAuthProvider.builder(authenticator).build();
+ * </pre>
+ *
+ * @author raccoonback
+ * @since 1.3.0
+ */
+public class GssCredentialAuthenticator implements SpnegoAuthenticator {
+
+	private final GSSCredential credential;
+
+	/**
+	 * Creates a new GssCredentialAuthenticator with the given GSSCredential.
+	 *
+	 * @param credential the GSSCredential to use for authentication
+	 */
+	public GssCredentialAuthenticator(GSSCredential credential) {
+		Objects.requireNonNull(credential, "GSSCredential cannot be null");
+		this.credential = credential;
+	}
+
+	/**
+	 * Creates a GSSContext for the specified service and remote host using the provided GSSCredential.
+	 * <p>
+	 * This method uses the pre-existing GSSCredential to create a GSSContext for SPNEGO
+	 * authentication. The service principal name is constructed as serviceName/remoteHost.
+	 * </p>
+	 *
+	 * @param serviceName the service name (e.g., "HTTP", "FTP")
+	 * @param remoteHost the remote host to authenticate with
+	 * @return the created GSSContext configured for SPNEGO authentication
+	 * @throws Exception if context creation fails
+	 */
+	@Override
+	public GSSContext createContext(String serviceName, String remoteHost) throws Exception {
+		GSSManager manager = GSSManager.getInstance();
+		GSSName serverName = manager.createName(serviceName + "/" + remoteHost, GSSName.NT_HOSTBASED_SERVICE);
+		GSSContext context = manager.createContext(
+				serverName,
+				new Oid("1.3.6.1.5.5.2"),
+				credential,
+				GSSContext.DEFAULT_LIFETIME
+		);
+		context.requestMutualAuth(true);
+		return context;
+	}
+}
\ No newline at end of file
diff --git a/reactor-netty-http/src/main/java/reactor/netty/http/client/HttpClient.java b/reactor-netty-http/src/main/java/reactor/netty/http/client/HttpClient.java
index e5b5f7527a..ea1e4dbfec 100644
--- a/reactor-netty-http/src/main/java/reactor/netty/http/client/HttpClient.java
+++ b/reactor-netty-http/src/main/java/reactor/netty/http/client/HttpClient.java
@@ -1698,6 +1698,20 @@ public final HttpClient wiretap(boolean enable) {
 		return super.wiretap(enable);
 	}
 
+	/**
+	 * Configure SPNEGO authentication for the HTTP client.
+	 *
+	 * @param spnegoAuthProvider the SPNEGO authentication provider
+	 * @return a new {@link HttpClient}
+	 * @since 1.3.0
+	 */
+	public final HttpClient spnego(SpnegoAuthProvider spnegoAuthProvider) {
+		Objects.requireNonNull(spnegoAuthProvider, "spnegoAuthProvider");
+		HttpClient dup = duplicate();
+		dup.configuration().spnegoAuthProvider = spnegoAuthProvider;
+		return dup;
+	}
+
 	static boolean isCompressing(HttpHeaders h) {
 		return h.contains(HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP, true)
 				|| h.contains(HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.BR, true);
diff --git a/reactor-netty-http/src/main/java/reactor/netty/http/client/HttpClientConfig.java b/reactor-netty-http/src/main/java/reactor/netty/http/client/HttpClientConfig.java
index e92e7da29b..05ac90e1c0 100644
--- a/reactor-netty-http/src/main/java/reactor/netty/http/client/HttpClientConfig.java
+++ b/reactor-netty-http/src/main/java/reactor/netty/http/client/HttpClientConfig.java
@@ -378,6 +378,7 @@ public WebsocketClientSpec websocketClientSpec() {
 	String uriStr;
 	Function<String, String> uriTagValue;
 	WebsocketClientSpec websocketClientSpec;
+	SpnegoAuthProvider spnegoAuthProvider;
 
 	HttpClientConfig(HttpConnectionProvider connectionProvider, Map<ChannelOption<?>, ?> options,
 			Supplier<? extends SocketAddress> remoteAddress) {
@@ -430,6 +431,7 @@ public WebsocketClientSpec websocketClientSpec() {
 		this.uriStr = parent.uriStr;
 		this.uriTagValue = parent.uriTagValue;
 		this.websocketClientSpec = parent.websocketClientSpec;
+		this.spnegoAuthProvider = parent.spnegoAuthProvider;
 	}
 
 	@Override
diff --git a/reactor-netty-http/src/main/java/reactor/netty/http/client/HttpClientConnect.java b/reactor-netty-http/src/main/java/reactor/netty/http/client/HttpClientConnect.java
index b0563c3e1d..47c7436861 100644
--- a/reactor-netty-http/src/main/java/reactor/netty/http/client/HttpClientConnect.java
+++ b/reactor-netty-http/src/main/java/reactor/netty/http/client/HttpClientConnect.java
@@ -75,6 +75,7 @@
  *
  * @author Stephane Maldini
  * @author Violeta Georgieva
+ * @author raccoonback
  */
 class HttpClientConnect extends HttpClient {
 
@@ -446,6 +447,16 @@ public Context currentContext() {
 		@Override
 		public void onStateChange(Connection connection, State newState) {
 			if (newState == HttpClientState.RESPONSE_RECEIVED) {
+				HttpClientOperations operations = connection.as(HttpClientOperations.class);
+				if (operations != null && handler.spnegoAuthProvider != null) {
+					if (shouldRetryWithSpnego(operations)) {
+						retryWithSpnego(operations);
+						return;
+					}
+
+					handler.spnegoAuthProvider.resetRetryCount();
+				}
+
 				sink.success(connection);
 				return;
 			}
@@ -459,6 +470,42 @@ public void onStateChange(Connection connection, State newState) {
 				    .subscribe(connection.disposeSubscriber());
 			}
 		}
+
+		/**
+		 * Determines if the current HTTP response requires a SPNEGO authentication retry.
+		 *
+		 * @param operations the HTTP client operations containing the response status and headers
+		 * @return {@code true} if SPNEGO re-authentication should be attempted, {@code false} otherwise
+		 */
+		private boolean shouldRetryWithSpnego(HttpClientOperations operations) {
+			int statusCode = operations.status().code();
+			HttpHeaders headers = operations.responseHeaders();
+
+			return handler.spnegoAuthProvider.isUnauthorized(statusCode, headers)
+				&& handler.spnegoAuthProvider.canRetry();
+		}
+
+		/**
+		 * Triggers a SPNEGO authentication retry by throwing a {@link SpnegoRetryException}.
+		 * <p>
+		 * The exception-based approach ensures that a completely new {@link HttpClientOperations}
+		 * instance is created, avoiding the "Status and headers already sent" error that would
+		 * occur if trying to reuse the existing connection.
+		 * </p>
+		 *
+		 * @param operations the current HTTP client operations that received the 401 response
+		 * @throws SpnegoRetryException always thrown to trigger the retry mechanism
+		 */
+		private void retryWithSpnego(HttpClientOperations operations) {
+			handler.spnegoAuthProvider.invalidateTokenHeader();
+			handler.spnegoAuthProvider.incrementRetryCount();
+
+			if (log.isDebugEnabled()) {
+				log.debug(format(operations.channel(), "Triggering SPNEGO re-authentication"));
+			}
+
+			sink.error(new SpnegoRetryException());
+		}
 	}
 
 	static final class HttpClientHandler extends SocketAddress
@@ -489,6 +536,8 @@ static final class HttpClientHandler extends SocketAddress
 		volatile boolean            shouldRetry;
 		volatile HttpHeaders        previousRequestHeaders;
 
+		SpnegoAuthProvider spnegoAuthProvider;
+
 		HttpClientHandler(HttpClientConfig configuration) {
 			this.method = configuration.method;
 			this.followRedirectPredicate = configuration.followRedirectPredicate;
@@ -526,6 +575,7 @@ static final class HttpClientHandler extends SocketAddress
 				this.fromURI = this.toURI = uriEndpointFactory.createUriEndpoint(configuration.uri, configuration.websocketClientSpec != null);
 			}
 			this.resourceUrl = toURI.toExternalForm();
+			this.spnegoAuthProvider = configuration.spnegoAuthProvider;
 		}
 
 		@Override
@@ -540,6 +590,19 @@ public SocketAddress get() {
 
 		@SuppressWarnings("ReferenceEquality")
 		Publisher<Void> requestWithBody(HttpClientOperations ch) {
+			if (spnegoAuthProvider != null) {
+				return spnegoAuthProvider.apply(ch, ch.address())
+					.then(
+							Mono.defer(
+									() -> Mono.from(requestWithBodyInternal(ch))
+							)
+					);
+			}
+
+			return requestWithBodyInternal(ch);
+		}
+
+		private Publisher<Void> requestWithBodyInternal(HttpClientOperations ch) {
 			try {
 				ch.resourceUrl = this.resourceUrl;
 				ch.responseTimeout = responseTimeout;
@@ -718,6 +781,9 @@ public boolean test(Throwable throwable) {
 				redirect(re.location);
 				return true;
 			}
+			if (throwable instanceof SpnegoRetryException) {
+				return true;
+			}
 			if (shouldRetry && AbortedException.isConnectionReset(throwable)) {
 				shouldRetry = false;
 				redirect(toURI.toString());
diff --git a/reactor-netty-http/src/main/java/reactor/netty/http/client/JaasAuthenticator.java b/reactor-netty-http/src/main/java/reactor/netty/http/client/JaasAuthenticator.java
new file mode 100644
index 0000000000..c5f23e2fb1
--- /dev/null
+++ b/reactor-netty-http/src/main/java/reactor/netty/http/client/JaasAuthenticator.java
@@ -0,0 +1,92 @@
+/*
+ * Copyright (c) 2025 VMware, Inc. or its affiliates, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package reactor.netty.http.client;
+
+import java.security.PrivilegedExceptionAction;
+import javax.security.auth.Subject;
+import javax.security.auth.login.LoginContext;
+import org.ietf.jgss.GSSContext;
+import org.ietf.jgss.GSSManager;
+import org.ietf.jgss.GSSName;
+import org.ietf.jgss.Oid;
+
+/**
+ * A JAAS-based Authenticator implementation for use with SPNEGO providers.
+ * <p>
+ * This authenticator performs a JAAS login using the specified context name and creates a GSSContext
+ * for SPNEGO authentication. It relies on JAAS configuration to obtain Kerberos credentials.
+ * </p>
+ * <p>
+ * The JAAS configuration should define a login context that acquires Kerberos credentials,
+ * typically using the Krb5LoginModule. The login context name provided to this authenticator
+ * must match the entry name in the JAAS configuration file.
+ * </p>
+ *
+ * <p>Example usage:</p>
+ * <pre>
+ *     JaasAuthenticator authenticator = new JaasAuthenticator("KerberosLogin");
+ *     SpnegoAuthProvider provider = SpnegoAuthProvider.builder(authenticator).build();
+ * </pre>
+ *
+ * @author raccoonback
+ * @since 1.3.0
+ */
+public class JaasAuthenticator implements SpnegoAuthenticator {
+
+	private final String loginContext;
+
+	/**
+	 * Creates a new JaasAuthenticator with the given context name.
+	 *
+	 * @param loginContext the JAAS login context name
+	 */
+	public JaasAuthenticator(String loginContext) {
+		this.loginContext = loginContext;
+	}
+
+	/**
+	 * Creates a GSSContext for the specified service and remote host using JAAS authentication.
+	 * <p>
+	 * This method performs a JAAS login, obtains the authenticated Subject, and creates
+	 * a GSSContext within the Subject's security context. The service principal name
+	 * is constructed as serviceName/remoteHost.
+	 * </p>
+	 *
+	 * @param serviceName the service name (e.g., "HTTP", "CIFS")
+	 * @param remoteHost the remote host to authenticate with
+	 * @return the created GSSContext configured for SPNEGO authentication
+	 * @throws Exception if JAAS login or context creation fails
+	 */
+	@Override
+	public GSSContext createContext(String serviceName, String remoteHost) throws Exception {
+		LoginContext lc = new LoginContext(loginContext);
+		lc.login();
+		Subject subject = lc.getSubject();
+
+		return Subject.doAs(subject, (PrivilegedExceptionAction<GSSContext>) () -> {
+			GSSManager manager = GSSManager.getInstance();
+			GSSName serverName = manager.createName(serviceName + "/" + remoteHost, GSSName.NT_HOSTBASED_SERVICE);
+			GSSContext context = manager.createContext(
+					serverName,
+					new Oid("1.3.6.1.5.5.2"), // SPNEGO
+					null,
+					GSSContext.DEFAULT_LIFETIME
+			);
+			context.requestMutualAuth(true);
+			return context;
+		});
+	}
+}
diff --git a/reactor-netty-http/src/main/java/reactor/netty/http/client/SpnegoAuthProvider.java b/reactor-netty-http/src/main/java/reactor/netty/http/client/SpnegoAuthProvider.java
new file mode 100644
index 0000000000..c5c169b504
--- /dev/null
+++ b/reactor-netty-http/src/main/java/reactor/netty/http/client/SpnegoAuthProvider.java
@@ -0,0 +1,315 @@
+/*
+ * Copyright (c) 2025 VMware, Inc. or its affiliates, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package reactor.netty.http.client;
+
+import static reactor.core.scheduler.Schedulers.boundedElastic;
+
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaders;
+import java.net.InetSocketAddress;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.ietf.jgss.GSSContext;
+import org.ietf.jgss.GSSException;
+import reactor.core.publisher.Mono;
+import reactor.util.Logger;
+import reactor.util.Loggers;
+
+/**
+ * Provides SPNEGO authentication for Reactor Netty HttpClient.
+ * <p>
+ * This provider is responsible for generating and attaching a SPNEGO (Kerberos) token
+ * to the HTTP Authorization header for outgoing requests, enabling single sign-on and
+ * secure authentication in enterprise environments.
+ * </p>
+ * <p>
+ * The provider supports authentication caching and retry mechanisms to handle token
+ * expiration and authentication failures gracefully. It can be configured with different
+ * service names, unauthorized status codes, and hostname resolution strategies.
+ * </p>
+ *
+ * <p>Basic usage with JAAS:</p>
+ * <pre>
+ *     SpnegoAuthProvider provider = SpnegoAuthProvider
+ *         .builder(new JaasAuthenticator("KerberosLogin"))
+ *         .build();
+ *
+ *     HttpClient client = HttpClient.create()
+ *         .spnego(provider);
+ * </pre>
+ *
+ * <p>Advanced configuration:</p>
+ * <pre>
+ *     SpnegoAuthProvider provider = SpnegoAuthProvider
+ *         .builder(new GssCredentialAuthenticator(credential))
+ *         .serviceName("CIFS")
+ *         .unauthorizedStatusCode(401)
+ *         .resolveCanonicalHostname(true)
+ *         .build();
+ * </pre>
+ *
+ * @author raccoonback
+ * @since 1.3.0
+ */
+public final class SpnegoAuthProvider {
+
+	private static final Logger log = Loggers.getLogger(SpnegoAuthProvider.class);
+	private static final String SPNEGO_HEADER = "Negotiate";
+
+	private final SpnegoAuthenticator authenticator;
+	private final int unauthorizedStatusCode;
+	private final String serviceName;
+	private final boolean resolveCanonicalHostname;
+
+	private final AtomicReference<String> verifiedAuthHeader = new AtomicReference<>();
+	private final AtomicInteger retryCount = new AtomicInteger(0);
+	private static final int MAX_RETRY_COUNT = 1;
+
+	/**
+	 * Constructs a new SpnegoAuthProvider with the specified configuration.
+	 * <p>
+	 * This constructor is private and should only be used by the {@link Builder}.
+	 * Use {@link #builder(SpnegoAuthenticator)} to create instances.
+	 * </p>
+	 *
+	 * @param authenticator the authenticator to use for SPNEGO authentication (must not be null)
+	 * @param unauthorizedStatusCode the HTTP status code that indicates authentication failure
+	 * @param serviceName the service name for constructing service principal names
+	 * @param resolveCanonicalHostname whether to resolve canonical hostnames for service principals
+	 */
+	private SpnegoAuthProvider(SpnegoAuthenticator authenticator, int unauthorizedStatusCode, String serviceName, boolean resolveCanonicalHostname) {
+		this.authenticator = authenticator;
+		this.unauthorizedStatusCode = unauthorizedStatusCode;
+		this.serviceName = serviceName;
+		this.resolveCanonicalHostname = resolveCanonicalHostname;
+	}
+
+	/**
+	 * Creates a new builder for configuring SPNEGO authentication provider.
+	 *
+	 * @param authenticator the authenticator to use for SPNEGO authentication
+	 * @return a new builder instance
+	 */
+	public static Builder builder(SpnegoAuthenticator authenticator) {
+		return new Builder(authenticator);
+	}
+
+	/**
+	 * Builder for creating SpnegoAuthProvider instances.
+	 */
+	public static final class Builder {
+		private final SpnegoAuthenticator authenticator;
+		private int unauthorizedStatusCode = 401;
+		private String serviceName = "HTTP";
+		private boolean resolveCanonicalHostname;
+
+		private Builder(SpnegoAuthenticator authenticator) {
+			this.authenticator = authenticator;
+		}
+
+		/**
+		 * Sets the HTTP status code that indicates authentication failure.
+		 *
+		 * @param statusCode the status code (default: 401)
+		 * @return this builder
+		 */
+		public Builder unauthorizedStatusCode(int statusCode) {
+			this.unauthorizedStatusCode = statusCode;
+			return this;
+		}
+
+		/**
+		 * Sets the service name for the service principal.
+		 *
+		 * @param serviceName the service name (default: "HTTP")
+		 * @return this builder
+		 */
+		public Builder serviceName(String serviceName) {
+			this.serviceName = serviceName;
+			return this;
+		}
+
+		/**
+		 * Sets whether to resolve canonical hostname.
+		 *
+		 * @param resolveCanonicalHostname true to resolve canonical hostname (default: false)
+		 * @return this builder
+		 */
+		public Builder resolveCanonicalHostname(boolean resolveCanonicalHostname) {
+			this.resolveCanonicalHostname = resolveCanonicalHostname;
+			return this;
+		}
+
+		/**
+		 * Builds the SpnegoAuthProvider instance.
+		 *
+		 * @return a new SpnegoAuthProvider
+		 */
+		public SpnegoAuthProvider build() {
+			return new SpnegoAuthProvider(authenticator, unauthorizedStatusCode, serviceName, resolveCanonicalHostname);
+		}
+	}
+
+	/**
+	 * Applies SPNEGO authentication to the given HTTP client request.
+	 * <p>
+	 * This method generates a SPNEGO token for the specified address and attaches it
+	 * as an Authorization header to the outgoing HTTP request.
+	 * </p>
+	 *
+	 * @param request the HTTP client request to authenticate
+	 * @param address the target server address (used for service principal)
+	 * @return a Mono that completes when the authentication is applied
+	 * @throws SpnegoAuthenticationException if login or token generation fails
+	 */
+	public Mono<Void> apply(HttpClientRequest request, InetSocketAddress address) {
+		String cachedToken = verifiedAuthHeader.get();
+		if (cachedToken != null) {
+			request.header(HttpHeaderNames.AUTHORIZATION, cachedToken);
+			return Mono.empty();
+		}
+
+		return Mono.fromCallable(() -> {
+				try {
+					String hostName = resolveHostName(address);
+					byte[] token = generateSpnegoToken(hostName);
+					String authHeader = SPNEGO_HEADER + " " + Base64.getEncoder().encodeToString(token);
+
+					verifiedAuthHeader.set(authHeader);
+					request.header(HttpHeaderNames.AUTHORIZATION, authHeader);
+					return token;
+				}
+				catch (Exception e) {
+					throw new SpnegoAuthenticationException("Failed to generate SPNEGO token", e);
+				}
+			})
+			.subscribeOn(boundedElastic())
+			.then();
+	}
+
+	/**
+	 * Resolves the hostname from the given socket address.
+	 * <p>
+	 * This method returns either the hostname or canonical hostname based on the
+	 * {@code resolveCanonicalHostname} configuration. When canonical hostname resolution
+	 * is enabled, it performs a reverse DNS lookup to get the fully qualified domain name.
+	 * </p>
+	 *
+	 * @param address the socket address to resolve hostname from
+	 * @return the resolved hostname (canonical if configured, otherwise standard hostname)
+	 */
+	private String resolveHostName(InetSocketAddress address) {
+		String hostName = address.getHostName();
+		if (resolveCanonicalHostname) {
+			hostName = address.getAddress().getCanonicalHostName();
+		}
+		return hostName;
+	}
+
+	/**
+	 * Generates a SPNEGO token for the given host name.
+	 * <p>
+	 * This method uses the authenticator to create a GSSContext and generate a SPNEGO token
+	 * for the specified service principal (HTTP/hostName).
+	 * </p>
+	 *
+	 * @param hostName the target server host name
+	 * @return the raw SPNEGO token bytes
+	 * @throws Exception if token generation fails
+	 */
+	private byte[] generateSpnegoToken(String hostName) throws Exception {
+		if (hostName == null || hostName.trim().isEmpty()) {
+			throw new IllegalArgumentException("Host name cannot be null or empty");
+		}
+
+		GSSContext context = null;
+		try {
+			context = authenticator.createContext(serviceName, hostName.trim());
+			return context.initSecContext(new byte[0], 0, 0);
+		}
+		finally {
+			if (context != null) {
+				try {
+					context.dispose();
+				}
+				catch (GSSException e) {
+					// Log but don't propagate disposal errors
+					if (log.isDebugEnabled()) {
+						log.debug("Failed to dispose GSSContext", e);
+					}
+				}
+			}
+		}
+	}
+
+	/**
+	 * Invalidates the cached authentication token.
+	 */
+	public void invalidateTokenHeader() {
+		this.verifiedAuthHeader.set(null);
+	}
+
+	/**
+	 * Checks if SPNEGO authentication retry is allowed.
+	 *
+	 * @return true if retry is allowed, false otherwise
+	 */
+	public boolean canRetry() {
+		return retryCount.get() < MAX_RETRY_COUNT;
+	}
+
+	/**
+	 * Increments the retry count for SPNEGO authentication attempts.
+	 */
+	public void incrementRetryCount() {
+		retryCount.incrementAndGet();
+	}
+
+	/**
+	 * Resets the retry count for SPNEGO authentication.
+	 */
+	public void resetRetryCount() {
+		retryCount.set(0);
+	}
+
+	/**
+	 * Checks if the response indicates an authentication failure that requires a new token.
+	 * <p>
+	 * This method checks both the status code and the WWW-Authenticate header to determine
+	 * if a new SPNEGO token needs to be generated.
+	 * </p>
+	 *
+	 * @param status the HTTP status code
+	 * @param headers the HTTP response headers
+	 * @return true if the response indicates an authentication failure
+	 */
+	public boolean isUnauthorized(int status, HttpHeaders headers) {
+		if (status != unauthorizedStatusCode) {
+			return false;
+		}
+
+		String header = headers.get(HttpHeaderNames.WWW_AUTHENTICATE);
+		if (header == null) {
+			return false;
+		}
+
+		return Arrays.stream(header.split(","))
+			.map(String::trim)
+			.anyMatch(auth -> auth.toLowerCase().startsWith(SPNEGO_HEADER.toLowerCase()));
+	}
+}
diff --git a/reactor-netty-http/src/main/java/reactor/netty/http/client/SpnegoAuthenticationException.java b/reactor-netty-http/src/main/java/reactor/netty/http/client/SpnegoAuthenticationException.java
new file mode 100644
index 0000000000..27e4923344
--- /dev/null
+++ b/reactor-netty-http/src/main/java/reactor/netty/http/client/SpnegoAuthenticationException.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright (c) 2025 VMware, Inc. or its affiliates, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package reactor.netty.http.client;
+
+/**
+ * Exception thrown when SPNEGO authentication fails.
+ *
+ * @author raccoonback
+ * @since 1.3.0
+ */
+public class SpnegoAuthenticationException extends RuntimeException {
+
+	public SpnegoAuthenticationException(String message, Throwable cause) {
+		super(message, cause);
+	}
+}
diff --git a/reactor-netty-http/src/main/java/reactor/netty/http/client/SpnegoAuthenticator.java b/reactor-netty-http/src/main/java/reactor/netty/http/client/SpnegoAuthenticator.java
new file mode 100644
index 0000000000..3924703599
--- /dev/null
+++ b/reactor-netty-http/src/main/java/reactor/netty/http/client/SpnegoAuthenticator.java
@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) 2025 VMware, Inc. or its affiliates, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package reactor.netty.http.client;
+
+import org.ietf.jgss.GSSContext;
+
+/**
+ * An abstraction for authentication logic used by SPNEGO providers.
+ * <p>
+ * Implementations are responsible for creating a GSSContext for the specified remote host.
+ * </p>
+ *
+ * @author raccoonback
+ * @since 1.3.0
+ */
+public interface SpnegoAuthenticator {
+
+	/**
+	 * Creates a GSSContext for the specified remote host.
+	 *
+	 * @param serviceName the service name (e.g., "HTTP", "FTP")
+	 * @param remoteHost the remote host to authenticate with
+	 * @return the created GSSContext
+	 * @throws Exception if context creation fails
+	 */
+	GSSContext createContext(String serviceName, String remoteHost) throws Exception;
+}
diff --git a/reactor-netty-http/src/main/java/reactor/netty/http/client/SpnegoRetryException.java b/reactor-netty-http/src/main/java/reactor/netty/http/client/SpnegoRetryException.java
new file mode 100644
index 0000000000..48182abff4
--- /dev/null
+++ b/reactor-netty-http/src/main/java/reactor/netty/http/client/SpnegoRetryException.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright (c) 2025 VMware, Inc. or its affiliates, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package reactor.netty.http.client;
+
+/**
+ * Exception thrown to trigger a retry when SPNEGO authentication fails with a 401 Unauthorized response.
+ *
+ * @author raccoonback
+ * @since 1.3.0
+ */
+final class SpnegoRetryException extends RuntimeException {
+
+	SpnegoRetryException() {
+		super("SPNEGO authentication requires retry");
+	}
+}
diff --git a/reactor-netty-http/src/test/java/reactor/netty/http/client/SpnegoAuthProviderTest.java b/reactor-netty-http/src/test/java/reactor/netty/http/client/SpnegoAuthProviderTest.java
new file mode 100644
index 0000000000..d5453dcdde
--- /dev/null
+++ b/reactor-netty-http/src/test/java/reactor/netty/http/client/SpnegoAuthProviderTest.java
@@ -0,0 +1,306 @@
+/*
+ * Copyright (c) 2025 VMware, Inc. or its affiliates, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package reactor.netty.http.client;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import io.netty.handler.codec.http.HttpHeaderNames;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.ietf.jgss.GSSContext;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+import reactor.netty.DisposableServer;
+import reactor.netty.http.server.HttpServer;
+import reactor.test.StepVerifier;
+
+class SpnegoAuthProviderTest {
+
+	private static final int TEST_PORT = 8080;
+
+	@Test
+	void negotiateSpnegoAuthenticationWithHttpClient() throws Exception {
+		DisposableServer server = HttpServer.create()
+				.port(TEST_PORT)
+				.route(routes -> routes
+						.get("/", (request, response) -> {
+							String authHeader = request.requestHeaders().get(HttpHeaderNames.AUTHORIZATION);
+							if (authHeader != null && authHeader.startsWith("Negotiate ")) {
+								return response.status(200).sendString(Mono.just("Authenticated"));
+							}
+							return response.status(401).sendString(Mono.just("Unauthorized"));
+						}))
+				.bindNow();
+
+		try {
+			GSSContext gssContext = mock(GSSContext.class);
+			SpnegoAuthenticator authenticator = mock(SpnegoAuthenticator.class);
+
+			given(gssContext.initSecContext(any(byte[].class), anyInt(), anyInt()))
+					.willReturn("spnego-negotiate-token".getBytes(StandardCharsets.UTF_8));
+			given(authenticator.createContext(anyString(), anyString()))
+					.willReturn(gssContext);
+
+			HttpClient client = HttpClient.create()
+					.port(TEST_PORT)
+					.spnego(
+							SpnegoAuthProvider.builder(authenticator)
+									.build()
+					)
+					.wiretap(true)
+					.disableRetry(true);
+
+			StepVerifier.create(
+							client.get()
+									.uri("/")
+									.responseContent()
+									.aggregate()
+									.asString()
+					)
+					.expectNext("Authenticated")
+					.verifyComplete();
+		}
+		finally {
+			server.disposeNow();
+		}
+	}
+
+	@Test
+	void automaticReauthenticateOn401Response() throws Exception {
+		AtomicInteger requestCount = new AtomicInteger(0);
+
+		DisposableServer server = HttpServer.create()
+				.port(0)
+				.route(routes -> routes
+						.get("/reauth", (request, response) -> {
+							String authHeader = request.requestHeaders().get(HttpHeaderNames.AUTHORIZATION);
+							int count = requestCount.incrementAndGet();
+
+							if (count == 1) {
+								return response.status(401)
+										.header("WWW-Authenticate", "Negotiate")
+										.sendString(Mono.just("Unauthorized"));
+							}
+							else if (authHeader != null && authHeader.startsWith("Negotiate ")) {
+								return response.status(200).sendString(Mono.just("Reauthenticated"));
+							}
+							return response.status(401).sendString(Mono.just("Failed"));
+						}))
+				.bindNow();
+
+		try {
+			GSSContext gssContext = mock(GSSContext.class);
+			SpnegoAuthenticator authenticator = mock(SpnegoAuthenticator.class);
+
+			given(gssContext.initSecContext(any(byte[].class), anyInt(), anyInt()))
+					.willReturn("spnego-reauth-token".getBytes(StandardCharsets.UTF_8));
+			given(authenticator.createContext(anyString(), anyString()))
+					.willReturn(gssContext);
+
+			HttpClient client = HttpClient.create()
+					.port(server.port())
+					.spnego(
+							SpnegoAuthProvider.builder(authenticator)
+									.build()
+					)
+					.wiretap(true)
+					.disableRetry(true);
+
+			StepVerifier.create(
+							client.get()
+									.uri("/reauth")
+									.responseContent()
+									.aggregate()
+									.asString()
+					)
+					.expectNext("Reauthenticated")
+					.verifyComplete();
+
+			verify(gssContext, times(2)).initSecContext(any(byte[].class), anyInt(), anyInt());
+		}
+		finally {
+			server.disposeNow();
+		}
+	}
+
+	@Test
+	void doesNotReauthenticateWhenMaxRetryReached() throws Exception {
+		AtomicInteger requestCount = new AtomicInteger(0);
+
+		DisposableServer server = HttpServer.create()
+				.port(0)
+				.route(routes -> routes
+						.get("/fail", (request, response) -> {
+							requestCount.incrementAndGet();
+							return response.status(401)
+									.header("WWW-Authenticate", "Negotiate")
+									.sendString(Mono.just("Always Unauthorized"));
+						}))
+				.bindNow();
+
+		try {
+			GSSContext gssContext = mock(GSSContext.class);
+			SpnegoAuthenticator authenticator = mock(SpnegoAuthenticator.class);
+
+			given(gssContext.initSecContext(any(byte[].class), anyInt(), anyInt()))
+					.willReturn("spnego-fail-token".getBytes(StandardCharsets.UTF_8));
+			given(authenticator.createContext(anyString(), anyString()))
+					.willReturn(gssContext);
+
+			HttpClient client = HttpClient.create()
+					.port(server.port())
+					.spnego(
+							SpnegoAuthProvider.builder(authenticator)
+									.build()
+					)
+					.wiretap(true)
+					.disableRetry(true);
+
+			StepVerifier.create(
+							client.get()
+									.uri("/fail")
+									.response()
+									.map(response -> response.status().code())
+					)
+					.expectNext(401)
+					.verifyComplete();
+
+			verify(gssContext, times(2)).initSecContext(any(byte[].class), anyInt(), anyInt());
+		}
+		finally {
+			server.disposeNow();
+		}
+	}
+
+	@Test
+	void doesNotReauthenticateWithoutWwwAuthenticateHeader() throws Exception {
+		DisposableServer server = HttpServer.create()
+				.port(0)
+				.route(routes -> routes
+						.get("/noheader", (request, response) ->
+								response.status(401).sendString(Mono.just("No WWW-Authenticate header"))))
+				.bindNow();
+
+		try {
+			GSSContext gssContext = mock(GSSContext.class);
+			SpnegoAuthenticator authenticator = mock(SpnegoAuthenticator.class);
+
+			given(gssContext.initSecContext(any(byte[].class), anyInt(), anyInt()))
+					.willReturn("spnego-token".getBytes(StandardCharsets.UTF_8));
+			given(authenticator.createContext(anyString(), anyString()))
+					.willReturn(gssContext);
+
+			HttpClient client = HttpClient.create()
+					.port(server.port())
+					.spnego(
+							SpnegoAuthProvider.builder(authenticator)
+									.build()
+					)
+					.wiretap(true)
+					.disableRetry(true);
+
+			StepVerifier.create(
+							client.get()
+									.uri("/noheader")
+									.response()
+									.map(response -> response.status().code())
+					)
+					.expectNext(401)
+					.verifyComplete();
+
+			verify(gssContext, times(1)).initSecContext(any(byte[].class), anyInt(), anyInt());
+		}
+		finally {
+			server.disposeNow();
+		}
+	}
+
+	@Test
+	void successfulAuthenticationResetsRetryCount() throws Exception {
+		AtomicInteger requestCount = new AtomicInteger(0);
+
+		DisposableServer server = HttpServer.create()
+				.port(0)
+				.route(routes -> routes
+						.get("/reset", (request, response) -> {
+							String authHeader = request.requestHeaders().get(HttpHeaderNames.AUTHORIZATION);
+							int count = requestCount.incrementAndGet();
+
+							if (count == 1) {
+								return response.status(401)
+										.header("WWW-Authenticate", "Negotiate")
+										.sendString(Mono.just("First 401"));
+							}
+							else if (authHeader != null && authHeader.startsWith("Negotiate ")) {
+								return response.status(200).sendString(Mono.just("Success"));
+							}
+							return response.status(401).sendString(Mono.just("Unexpected"));
+						}))
+				.bindNow();
+
+		try {
+			GSSContext gssContext = mock(GSSContext.class);
+			SpnegoAuthenticator authenticator = mock(SpnegoAuthenticator.class);
+
+			given(gssContext.initSecContext(any(byte[].class), anyInt(), anyInt()))
+					.willReturn("spnego-reset-token".getBytes(StandardCharsets.UTF_8));
+			given(authenticator.createContext(anyString(), anyString()))
+					.willReturn(gssContext);
+
+			SpnegoAuthProvider provider = SpnegoAuthProvider.builder(authenticator)
+					.build();
+
+			HttpClient client = HttpClient.create()
+					.port(server.port())
+					.spnego(provider)
+					.wiretap(true)
+					.disableRetry(true);
+
+			StepVerifier.create(
+							client.get()
+									.uri("/reset")
+									.responseContent()
+									.aggregate()
+									.asString()
+					)
+					.expectNext("Success")
+					.verifyComplete();
+
+			requestCount.set(0);
+
+			StepVerifier.create(
+							client.get()
+									.uri("/reset")
+									.responseContent()
+									.aggregate()
+									.asString()
+					)
+					.expectNext("Success")
+					.verifyComplete();
+
+			verify(gssContext, times(3)).initSecContext(any(byte[].class), anyInt(), anyInt());
+		}
+		finally {
+			server.disposeNow();
+		}
+	}
+}