diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/Login.java b/zookeeper-server/src/main/java/org/apache/zookeeper/Login.java index 2c483a5f74b..5ed31c84a59 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/Login.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/Login.java @@ -69,8 +69,14 @@ public class Login { private static final long MIN_TIME_BEFORE_RELOGIN = Long.getLong( MIN_TIME_BEFORE_RELOGIN_CONFIG_KEY, DEFAULT_MIN_TIME_BEFORE_RELOGIN); + private static final long DEFAULT_SHUTDOWN_TIMEOUT = 5 * 1000L; + public static final String SHUTDOWN_TIMEOUT_CONFIG_KEY = "zookeeper.kerberos.shutdownTimeoutMs"; + private static final long SHUTDOWN_TIMEOUT = Math.max(1L, + Long.getLong(SHUTDOWN_TIMEOUT_CONFIG_KEY, DEFAULT_SHUTDOWN_TIMEOUT)); + private Subject subject = null; private Thread t = null; + private volatile boolean shutdownRequested = false; private boolean isKrbTicket = false; private boolean isUsingTicketCache = false; @@ -132,7 +138,7 @@ public Login(final String loginContextName, Supplier callbackHa t = new Thread(new Runnable() { public void run() { LOG.info("TGT refresh thread started."); - while (true) { // renewal thread's main loop. if it exits from here, thread will exit. + while (!shutdownRequested) { // renewal thread's main loop. if it exits from here, thread will exit. KerberosTicket tgt = getTGT(); long now = Time.currentWallTime(); long nextRefresh; @@ -219,6 +225,9 @@ public void run() { int retry = 1; while (retry >= 0) { try { + if (Thread.currentThread().isInterrupted() || shutdownRequested) { + return; + } LOG.debug("running ticket cache refresh command: {} {}", cmd, kinitArgs); Shell.execCommand(cmd, kinitArgs); break; @@ -232,6 +241,9 @@ public void run() { LOG.error("Interrupted while renewing TGT, exiting Login thread"); return; } + } else if (Thread.currentThread().isInterrupted() || shutdownRequested) { + LOG.info("Shutdown requested while renewing TGT, exiting Login thread"); + return; } else { LOG.warn( "Could not renew TGT due to problem running shell command: '{} {}'." @@ -244,6 +256,10 @@ public void run() { } } } + // reLogin() is not interruptible, do not enter it while shutting down + if (shutdownRequested) { + break; + } try { int retry = 1; while (retry >= 0) { @@ -294,12 +310,17 @@ public void startThreadIfNeeded() { } public void shutdown() { + shutdownRequested = true; if ((t != null) && (t.isAlive())) { t.interrupt(); try { - t.join(); + t.join(SHUTDOWN_TIMEOUT); + if (t.isAlive()) { + LOG.warn("TGT renewal thread did not exit within {} ms and is being abandoned.", SHUTDOWN_TIMEOUT); + } } catch (InterruptedException e) { LOG.warn("error while waiting for Login thread to shutdown.", e); + Thread.currentThread().interrupt(); } } } diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/Shell.java b/zookeeper-server/src/main/java/org/apache/zookeeper/Shell.java index f780ff98556..6c35ee14b6c 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/Shell.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/Shell.java @@ -235,6 +235,8 @@ public void run() { errThread.join(); } catch (InterruptedException ie) { LOG.warn("Interrupted while reading the error stream", ie); + // join() cleared the interrupt status, restore it for the caller + Thread.currentThread().interrupt(); } completed.set(true); //the timeout thread handling @@ -243,6 +245,8 @@ public void run() { throw new ExitCodeException(exitCode, errMsg.toString()); } } catch (InterruptedException ie) { + // waitFor() cleared the interrupt status, restore it for the caller + Thread.currentThread().interrupt(); throw new IOException(ie.toString()); } finally { if ((timeOutTimer != null) && !timedOut.get()) { diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/KerberosTicketRenewalTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/KerberosTicketRenewalTest.java index d0f52b152af..f8d8e66c0bb 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/KerberosTicketRenewalTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/KerberosTicketRenewalTest.java @@ -24,6 +24,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertTimeout; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.File; import java.io.FileWriter; @@ -72,6 +73,7 @@ public static void setupClass() throws Exception { // by default, we should wait at least 1 minute between subsequent TGT renewals. // changing it to 500ms. System.setProperty(Login.MIN_TIME_BEFORE_RELOGIN_CONFIG_KEY, "500"); + System.setProperty(Login.SHUTDOWN_TIMEOUT_CONFIG_KEY, "200"); testTempDir = ClientBase.createTmpDir(); startMiniKdcAndAddPrincipal(); @@ -99,6 +101,7 @@ public static void setupClass() throws Exception { @AfterAll public static void tearDownClass() { System.clearProperty(Login.MIN_TIME_BEFORE_RELOGIN_CONFIG_KEY); + System.clearProperty(Login.SHUTDOWN_TIMEOUT_CONFIG_KEY); System.clearProperty("java.security.auth.login.config"); stopMiniKdc(); if (testTempDir != null) { @@ -125,6 +128,8 @@ private static class TestableKerberosLogin extends Login { private AtomicBoolean refreshFailed = new AtomicBoolean(false); private CountDownLatch continueRefreshThread = new CountDownLatch(1); + private volatile boolean hangUninterruptibly = false; + private final CountDownLatch hungThreadLatch = new CountDownLatch(1); public TestableKerberosLogin() throws LoginException { super(JAAS_CONFIG_SECTION, () -> { @@ -136,10 +141,29 @@ public TestableKerberosLogin() throws LoginException { protected void sleepBeforeRetryFailedRefresh() throws InterruptedException { LOG.info("sleep started due to failed refresh"); refreshFailed.set(true); - continueRefreshThread.await(20, TimeUnit.SECONDS); + if (hangUninterruptibly) { + // stands in for LoginContext.login(), which ignores interrupts + while (hungThreadLatch.getCount() > 0) { + try { + hungThreadLatch.await(); + } catch (InterruptedException deliberatelyIgnored) { + // ignored on purpose + } + } + } else { + continueRefreshThread.await(20, TimeUnit.SECONDS); + } LOG.info("sleep due to failed refresh finished"); } + public void hangUninterruptiblyOnFailedRefresh() { + hangUninterruptibly = true; + } + + public void releaseHungThread() { + hungThreadLatch.countDown(); + } + public void assertRefreshFailsEventually(Duration timeout) { assertEventually(timeout, () -> refreshFailed.get()); } @@ -200,6 +224,26 @@ public void shouldRecoverIfKerberosNotAvailableForSomeTime() throws Exception { } + @Test + public void shouldNotBlockForeverWhenRenewalThreadDoesNotExit() throws Exception { + login = new TestableKerberosLogin(); + login.hangUninterruptiblyOnFailedRefresh(); + login.startThreadIfNeeded(); + + // the first renewal already fails, and the thread then parks in a call that + // ignores interrupts + stopMiniKdc(); + login.assertRefreshFailsEventually(Duration.ofSeconds(15)); + + try { + assertTimeoutPreemptively(Duration.ofSeconds(10), () -> login.shutdown()); + } finally { + startMiniKdcAndAddPrincipal(); + login.releaseHungThread(); + } + } + + private void assertPrincipalLoggedIn() { assertEquals(PRINCIPAL, login.getUserName()); assertNotNull(login.getSubject()); diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/ShellTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/ShellTest.java new file mode 100644 index 00000000000..52513419f72 --- /dev/null +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/ShellTest.java @@ -0,0 +1,86 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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 org.apache.zookeeper; + +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import java.io.IOException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledOnOs; +import org.junit.jupiter.api.condition.OS; + +public class ShellTest { + + /** + * Interrupting a thread running an external command must not swallow the interrupt. + * The interrupt does not abort the child process, it only has to survive into the + * caller, which is what {@link Login} relies on to notice it was asked to stop. + */ + @Test + @DisabledOnOs(OS.WINDOWS) + public void shouldRestoreInterruptStatusWhenInterruptedWhileWaitingForProcess() throws Exception { + CountDownLatch started = new CountDownLatch(1); + CountDownLatch finished = new CountDownLatch(1); + AtomicBoolean interruptedAfterwards = new AtomicBoolean(false); + AtomicReference expectedFailure = new AtomicReference<>(); + AtomicReference unexpectedFailure = new AtomicReference<>(); + + Thread runner = new Thread(() -> { + started.countDown(); + try { + // streams closed up front, so execCommand() reaches Process.waitFor() + // while the child is still alive + Shell.execCommand("sh", "-c", "exec >/dev/null 2>&1; sleep 2"); + } catch (IOException e) { + expectedFailure.set(e); + } catch (Throwable t) { + unexpectedFailure.set(t); + } finally { + interruptedAfterwards.set(Thread.currentThread().isInterrupted()); + finished.countDown(); + } + }); + runner.start(); + + assertTrue(started.await(10, TimeUnit.SECONDS), "runner thread did not start"); + + // waitFor() is the only blocking call left, wait for the thread to park in it + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (runner.getState() != Thread.State.WAITING) { + assertTrue(System.nanoTime() < deadline, "runner thread never reached Process.waitFor()"); + Thread.onSpinWait(); + } + runner.interrupt(); + + assertTrue(finished.await(30, TimeUnit.SECONDS), "execCommand did not return"); + runner.join(); + + assertNull(unexpectedFailure.get(), "unexpected failure in runner thread"); + assertNotNull(expectedFailure.get(), "execCommand was expected to fail with an IOException"); + assertTrue(interruptedAfterwards.get(), + "execCommand must leave the interrupt status set, otherwise the caller never " + + "learns that it was asked to stop"); + } + +}