wcarlson5 commented on a change in pull request #9978:
URL: https://github.com/apache/kafka/pull/9978#discussion_r565815347
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -1208,24 +1227,28 @@ private Thread shutdownHelper(final boolean error) {
}
private boolean close(final long timeoutMs) {
- if (state == State.ERROR) {
- log.info("Streams client is already in the terminal state ERROR,
all resources are closed and the client has stopped.");
+ if (state == State.ERROR || state == State.NOT_RUNNING) {
+ log.info("Streams client is already in the terminal {} state, all
resources are closed and the client has stopped.", state);
return true;
}
- if (state == State.PENDING_ERROR) {
- log.info("Streams client is in PENDING_ERROR, all resources are
being closed and the client will be stopped.");
- if (waitOnState(State.ERROR, timeoutMs)) {
+ if (state == State.PENDING_ERROR || state == State.PENDING_SHUTDOWN) {
+ log.info("Streams client is in {}, all resources are being closed
and the client will be stopped.", state);
+ if (state == State.PENDING_ERROR && waitOnState(State.ERROR,
timeoutMs)) {
log.info("Streams client stopped to ERROR completely");
return true;
+ } else if (state == State.PENDING_SHUTDOWN &&
waitOnState(State.NOT_RUNNING, timeoutMs)) {
+ log.info("Streams client stopped to NOT_RUNNING completely");
+ return true;
} else {
- log.info("Streams client cannot transition to ERROR completely
within the timeout");
+ log.warn("Streams client cannot transition to {}} completely
within the timeout", state);
Review comment:
the state here doesn't make the log make sense. If the state is
`PENDING_ERROR` then the log should say ERROR
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java
##########
@@ -133,6 +152,72 @@ private void configurePermissions(final File file) {
}
}
+ /**
+ * @return true if the state directory was successfully locked
+ */
+ private boolean lockStateDirectory() {
+ final File lockFile = new File(stateDir, LOCK_FILE_NAME);
+ try {
+ stateDirLockChannel = FileChannel.open(lockFile.toPath(),
StandardOpenOption.CREATE, StandardOpenOption.WRITE);
+ stateDirLock = tryLock(stateDirLockChannel);
+ } catch (final IOException e) {
+ log.error("Unable to lock the state directory due to unexpected
exception", e);
+ throw new ProcessorStateException("Failed to lock the state
directory during startup", e);
+ }
+
+ return stateDirLock != null;
+ }
+
+ public UUID initializeProcessId() {
Review comment:
since it doesn't seem that we need to be very thrifty with space for
this file would it make sense to write it in a more friendly format that would
be easier to maintain? i.e. json or something, we are giving it a version
number...
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -1208,24 +1227,28 @@ private Thread shutdownHelper(final boolean error) {
}
private boolean close(final long timeoutMs) {
- if (state == State.ERROR) {
- log.info("Streams client is already in the terminal state ERROR,
all resources are closed and the client has stopped.");
+ if (state == State.ERROR || state == State.NOT_RUNNING) {
Review comment:
I think this change makes a lot of sense. I don't think it changes the
final behavior besides avoiding extra state change rejections from the logs,
but it looks like they are replaced.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java
##########
@@ -133,6 +152,72 @@ private void configurePermissions(final File file) {
}
}
+ /**
+ * @return true if the state directory was successfully locked
+ */
+ private boolean lockStateDirectory() {
+ final File lockFile = new File(stateDir, LOCK_FILE_NAME);
+ try {
+ stateDirLockChannel = FileChannel.open(lockFile.toPath(),
StandardOpenOption.CREATE, StandardOpenOption.WRITE);
+ stateDirLock = tryLock(stateDirLockChannel);
Review comment:
Is there any case where we might want to release the lock of this state
directory? It looks like we just hold it
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
