ableegoldman commented on a change in pull request #9978:
URL: https://github.com/apache/kafka/pull/9978#discussion_r566550049
##########
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() {
+ if (!hasPersistentStores) {
+ return UUID.randomUUID();
+ }
+
+ if (!lockStateDirectory()) {
+ log.error("Unable to obtain lock as state directory is already
locked by another process");
+ throw new StreamsException("Unable to initialize state, this can
happen if multiple instances of " +
+ "Kafka Streams are running in the
same state directory");
+ }
+
+ final File processFile = new File(stateDir, PROCESS_FILE_NAME);
+ try {
+ if (processFile.exists()) {
+ try (final BufferedReader reader =
Files.newBufferedReader(processFile.toPath())) {
+ // only field in version 0 is the UUID
+ final int version = Integer.parseInt(reader.readLine());
+ if (version > 0) {
Review comment:
Oh yeah definitely, thanks
----------------------------------------------------------------
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
