Skip to content

MINOR: prevent cleanup() from being called while Streams is still shutting down #10666

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ableegoldman merged 3 commits into from
May 13, 2021
Merged

MINOR: prevent cleanup() from being called while Streams is still shutting down #10666

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
373a93b
throw if still shutting down and add test
ableegoldman May 11, 2021
ca42ba9
unused imports
ableegoldman May 11, 2021
49ceac2
test fix
ableegoldman May 11, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1444,7 +1444,7 @@ public synchronized boolean close(final Duration timeout) throws IllegalArgument
* @throws StreamsException if cleanup failed
*/
public void cleanUp() {
if (isRunningOrRebalancing()) {
if (!(state == State.CREATED || state == State.NOT_RUNNING || state == State.ERROR)) {
throw new IllegalStateException("Cannot clean up while running.");
}
stateDirectory.clean();
Expand Down
54 changes: 21 additions & 33 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -351,7 +351,7 @@ public void close() {

public synchronized void clean() {
try {
cleanRemovedTasksCalledByUser();
cleanStateAndTaskDirectoriesCalledByUser();
} catch (final Exception e) {
throw new StreamsException(e);
}
Expand Down Expand Up @@ -413,43 +413,31 @@ private void cleanRemovedTasksCalledByCleanerThread(final long cleanupDelayMs) {
}
}

private void cleanRemovedTasksCalledByUser() throws Exception {
private void cleanStateAndTaskDirectoriesCalledByUser() throws Exception {
if (!lockedTasksToOwner.isEmpty()) {
log.warn("Found some still-locked task directories when user requested to cleaning up the state, "
+ "since Streams is not running any more these will be ignored to complete the cleanup");
}
final AtomicReference<Exception> firstException = new AtomicReference<>();
for (final File taskDir : listAllTaskDirectories()) {
final String dirName = taskDir.getName();
final TaskId id = TaskId.parseTaskDirectoryName(dirName, null);
if (!lockedTasksToOwner.containsKey(id)) {
try {
if (lock(id)) {
log.info("{} Deleting state directory {} for task {} as user calling cleanup.",
logPrefix(), dirName, id);
Utils.delete(taskDir);
} else {
log.warn("{} Could not get lock for state directory {} for task {} as user calling cleanup.",
logPrefix(), dirName, id);
}
} catch (final OverlappingFileLockException | IOException exception) {
log.error(
String.format("%s Failed to delete state directory %s for task %s with exception:",
logPrefix(), dirName, id),
exception
);
firstException.compareAndSet(null, exception);
} finally {
try {
unlock(id);
// for manual user call, stream threads are not running so it is safe to delete
// the whole directory
Utils.delete(taskDir);
} catch (final IOException exception) {
log.error(
String.format("%s Failed to release lock on state directory %s for task %s with exception:",
logPrefix(), dirName, id),
exception
);
firstException.compareAndSet(null, exception);
}
try {
log.info("{} Deleting state directory {} for task {} as user calling cleanup.",
logPrefix(), dirName, id);

if (lockedTasksToOwner.containsKey(id)) {
log.warn("{} Task {} in state directory {} was still locked by {}",
logPrefix(), dirName, id, lockedTasksToOwner.get(id));
}
Utils.delete(taskDir);

@guozhangwang guozhangwang May 13, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah that's a nice one: we are sure all threads are gone now so there's no need to lock :)

} catch (final IOException exception) {
log.error(
String.format("%s Failed to delete state directory %s for task %s with exception:",
logPrefix(), dirName, id),
exception
);
firstException.compareAndSet(null, exception);
}
}
final Exception exception = firstException.get();
Expand Down
64 changes: 42 additions & 22 deletions streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.KafkaStreams.State;
import org.apache.kafka.streams.errors.StreamsNotStartedException;
import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
import org.apache.kafka.streams.errors.TopologyException;
Expand Down Expand Up @@ -96,6 +97,7 @@
import static org.apache.kafka.streams.state.QueryableStoreTypes.keyValueStore;
import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName;
import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitForApplicationState;
import static org.apache.kafka.test.TestUtils.waitForCondition;
import static org.easymock.EasyMock.anyInt;
import static org.easymock.EasyMock.anyLong;
import static org.easymock.EasyMock.anyObject;
Expand Down Expand Up @@ -235,8 +237,8 @@ private void prepareStreams() throws Exception {

EasyMock.expect(StreamThread.eosEnabled(anyObject(StreamsConfig.class))).andReturn(false).anyTimes();
EasyMock.expect(StreamThread.processingMode(anyObject(StreamsConfig.class))).andReturn(StreamThread.ProcessingMode.AT_LEAST_ONCE).anyTimes();
EasyMock.expect(streamThreadOne.getId()).andReturn(0L).anyTimes();
EasyMock.expect(streamThreadTwo.getId()).andReturn(1L).anyTimes();
EasyMock.expect(streamThreadOne.getId()).andReturn(1L).anyTimes();
EasyMock.expect(streamThreadTwo.getId()).andReturn(2L).anyTimes();
prepareStreamThread(streamThreadOne, 1, true);
prepareStreamThread(streamThreadTwo, 2, false);

Expand Down Expand Up @@ -298,7 +300,9 @@ private void prepareStreams() throws Exception {
);
}

private void prepareStreamThread(final StreamThread thread, final int threadId, final boolean terminable) throws Exception {
private void prepareStreamThread(final StreamThread thread,
Comment thread
ableegoldman marked this conversation as resolved.
final int threadId,
final boolean terminable) throws Exception {
final AtomicReference<StreamThread.State> state = new AtomicReference<>(StreamThread.State.CREATED);
EasyMock.expect(thread.state()).andAnswer(state::get).anyTimes();

Expand Down Expand Up @@ -351,19 +355,21 @@ private void prepareStreamThread(final StreamThread thread, final int threadId,
producer.close();
}
state.set(StreamThread.State.DEAD);

threadStatelistenerCapture.getValue().onChange(thread, StreamThread.State.PENDING_SHUTDOWN, StreamThread.State.RUNNING);
threadStatelistenerCapture.getValue().onChange(thread, StreamThread.State.DEAD, StreamThread.State.PENDING_SHUTDOWN);
return null;
}).anyTimes();
EasyMock.expect(thread.isRunning()).andReturn(state.get() == StreamThread.State.RUNNING).anyTimes();
thread.join();
if (terminable)
if (terminable) {
EasyMock.expectLastCall().anyTimes();
else
} else {
EasyMock.expectLastCall().andAnswer(() -> {
Thread.sleep(50L);
Thread.sleep(2000L);
return null;
}).anyTimes();
}

EasyMock.expect(thread.activeTasks()).andStubReturn(emptyList());
EasyMock.expect(thread.allTasks()).andStubReturn(Collections.emptyMap());
Expand All @@ -387,7 +393,7 @@ public void stateShouldTransitToRunningIfNonDeadThreadsBackToRunning() throws In

streams.start();

TestUtils.waitForCondition(
waitForCondition(
() -> streamsStateListener.numChanges == 2,
"Streams never started.");
Assert.assertEquals(KafkaStreams.State.RUNNING, streams.state());
Expand Down Expand Up @@ -439,7 +445,7 @@ public void stateShouldTransitToRunningIfNonDeadThreadsBackToRunning() throws In

streams.close();

TestUtils.waitForCondition(
waitForCondition(
() -> streamsStateListener.numChanges == 6,
"Streams never closed.");
Assert.assertEquals(KafkaStreams.State.NOT_RUNNING, streams.state());
Expand All @@ -453,7 +459,7 @@ public void shouldCleanupResourcesOnCloseWithoutPreviousStart() throws Exception
final KafkaStreams streams = new KafkaStreams(builder.build(), props, supplier, time);
streams.close();

TestUtils.waitForCondition(
waitForCondition(
() -> streams.state() == KafkaStreams.State.NOT_RUNNING,
"Streams never stopped.");

Expand All @@ -476,26 +482,26 @@ public void testStateThreadClose() throws Exception {
assertEquals(streams.state(), KafkaStreams.State.CREATED);

streams.start();
TestUtils.waitForCondition(
waitForCondition(
() -> streams.state() == KafkaStreams.State.RUNNING,
"Streams never started.");

for (int i = 0; i < NUM_THREADS; i++) {
final StreamThread tmpThread = streams.threads.get(i);
tmpThread.shutdown();
TestUtils.waitForCondition(() -> tmpThread.state() == StreamThread.State.DEAD,
waitForCondition(() -> tmpThread.state() == StreamThread.State.DEAD,
"Thread never stopped.");
streams.threads.get(i).join();
}
TestUtils.waitForCondition(
waitForCondition(
() -> streams.localThreadsMetadata().stream().allMatch(t -> t.threadState().equals("DEAD")),
"Streams never stopped"
);
} finally {
streams.close();
}

TestUtils.waitForCondition(
waitForCondition(
() -> streams.state() == KafkaStreams.State.NOT_RUNNING,
"Streams never stopped.");

Expand All @@ -511,25 +517,25 @@ public void testStateGlobalThreadClose() throws Exception {

try {
streams.start();
TestUtils.waitForCondition(
waitForCondition(
() -> streams.state() == KafkaStreams.State.RUNNING,
"Streams never started.");

final GlobalStreamThread globalStreamThread = streams.globalStreamThread;
globalStreamThread.shutdown();
TestUtils.waitForCondition(
waitForCondition(
() -> globalStreamThread.state() == GlobalStreamThread.State.DEAD,
"Thread never stopped.");
globalStreamThread.join();
TestUtils.waitForCondition(
waitForCondition(
() -> streams.state() == KafkaStreams.State.PENDING_ERROR,
"Thread never stopped."
);
} finally {
streams.close();
}

TestUtils.waitForCondition(
waitForCondition(
() -> streams.state() == KafkaStreams.State.ERROR,
"Thread never stopped."
);
Expand Down Expand Up @@ -568,7 +574,7 @@ public void shouldAddThreadWhenRunning() throws InterruptedException {
final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time);
streams.start();
final int oldSize = streams.threads.size();
TestUtils.waitForCondition(() -> streams.state() == KafkaStreams.State.RUNNING, 15L, "wait until running");
waitForCondition(() -> streams.state() == KafkaStreams.State.RUNNING, 15L, "wait until running");
assertThat(streams.addStreamThread(), equalTo(Optional.of("processId-StreamThread-" + 2)));
assertThat(streams.threads.size(), equalTo(oldSize + 1));
}
Expand Down Expand Up @@ -616,7 +622,7 @@ public void shouldRemoveThread() throws InterruptedException {
final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time);
streams.start();
final int oldSize = streams.threads.size();
TestUtils.waitForCondition(() -> streams.state() == KafkaStreams.State.RUNNING, 15L,
waitForCondition(() -> streams.state() == KafkaStreams.State.RUNNING, 15L,
"Kafka Streams client did not reach state RUNNING");
assertThat(streams.removeStreamThread(), equalTo(Optional.of("processId-StreamThread-" + 1)));
assertThat(streams.threads.size(), equalTo(oldSize - 1));
Expand Down Expand Up @@ -707,7 +713,7 @@ public void shouldAllowCleanupBeforeStartAndAfterClose() {
public void shouldThrowOnCleanupWhileRunning() throws InterruptedException {
final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time);
streams.start();
TestUtils.waitForCondition(
waitForCondition(
() -> streams.state() == KafkaStreams.State.RUNNING,
"Streams never started.");

Expand All @@ -719,6 +725,20 @@ public void shouldThrowOnCleanupWhileRunning() throws InterruptedException {
}
}

@Test
public void shouldThrowOnCleanupWhileShuttingDown() throws InterruptedException {
final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time);
streams.start();
waitForCondition(
() -> streams.state() == KafkaStreams.State.RUNNING,
"Streams never started.");

streams.close(Duration.ZERO);
assertThat(streams.state() == State.PENDING_SHUTDOWN, equalTo(true));
assertThrows(IllegalStateException.class, streams::cleanUp);
assertThat(streams.state() == State.PENDING_SHUTDOWN, equalTo(true));
}

@Test
public void shouldNotGetAllTasksWhenNotRunning() throws InterruptedException {
try (final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time)) {
Expand Down Expand Up @@ -1027,13 +1047,13 @@ public void shouldTransitToRunningWithGlobalOnlyTopology() throws InterruptedExc
assertEquals(streams.state(), KafkaStreams.State.CREATED);

streams.start();
TestUtils.waitForCondition(
waitForCondition(
() -> streams.state() == KafkaStreams.State.RUNNING,
"Streams never started, state is " + streams.state());

streams.close();

TestUtils.waitForCondition(
waitForCondition(
() -> streams.state() == KafkaStreams.State.NOT_RUNNING,
"Streams never stopped.");
}
Expand Down
4 changes: 2 additions & 2 deletions ...ams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,8 +170,8 @@ public void shouldSurviveWithOneTaskAsStandby() throws Exception {
// Wait for the record to be processed
assertTrue(instanceLatch.await(15, TimeUnit.SECONDS));

streamInstanceOne.close(Duration.ZERO);
streamInstanceTwo.close(Duration.ZERO);
Comment on lines -173 to -174

@ableegoldman ableegoldman May 13, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This test failing does give me some small pause on just throwing an IllegalStateException in case of PENDING_SHUTDOWN/ERROR -- do we think many users follow this pattern of calling the non-blocking close and then immediately calling cleanUp()? Maybe the better question is, should we support this, or is it alright to just enforce that users wait for Streams to shut down before invoking cleanUp -- if we do want to support this, we could just signal to the shutdown thread that it should re-invoke cleanUp once Streams finished shutting down, instead of just throwing an IllegalStateException. Not sure if it's really worth it or not, though -- thoughts @guozhangwang @wcarlson5 ?

@wcarlson5 wcarlson5 May 13, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How hard could it be to have cleanUp() check set a flag for the shutdown thread to call clean up after it's done? We want to make sure the shutdown thread checks after it changes the state to NOT_RUNNING or ERROR.

I think this a much more flexible pattern that would be good to support if it's running or rebalancing we would still want to throw and IllegalStateException or maybe IllegalOperationException. But either way it would be good to test when stream is in a PENDING state. But I think you would have to make that a unit test

@ableegoldman ableegoldman May 13, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I already have a unit test in this PR for the PENDING case, I should be able to adapt that to the new behavior without much trouble, if we want to change it. I also don't think it would be that difficult to implement this behavior -- I guess it was more a question of whether we should handle this semi-transparently or just tell the user to be patient and try again after the close. For example, at the moment close() will launch the shutdown helper as a daemon thread, so if the user does a non-blocking close followed by cleanUp and that's the end of the app, most likely the clean up will not occur (nor will the app complete its shutdown)

@ableegoldman ableegoldman May 13, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess another alternative is just to have cleanUp block until Streams finishes shutting down, but I don't think we want to make it essentially block on the results of close(), at least not without giving users a way to supply a timeout, and I don't think it's worth doing an entire KIP just to add an overload with one

@wcarlson5 wcarlson5 May 13, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think if they call a non blocking close then clean up then later notice the leftovers that is not a great experience and if can prevent it we should. I agree probably not worth a KIP though

@guozhangwang guozhangwang May 13, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

At the moment, if user calls a close() with no blocking and then cleanUp, the latter cleanUp could become a no-op because the locks are not released yet, and user won't know; whereas now it would throw an exception saying that close is not done yet, right? Plus our javadoc of cleanUp states that

     * Do a clean up of the local {@link StateStore} directory ({@link StreamsConfig#STATE_DIR_CONFIG}) by deleting all
     * data with regard to the {@link StreamsConfig#APPLICATION_ID_CONFIG application ID}.
     * <p>
     * May only be called either before this {@code KafkaStreams} instance is {@link #start() started} or after the
     * instance is {@link #close() closed}.

Hence I think the new behavior is not really a regression but really a fix forward.

streamInstanceOne.close();
streamInstanceTwo.close();

streamInstanceOne.cleanUp();
streamInstanceTwo.cleanUp();
Expand Down