ableegoldman commented on a change in pull request #10342:
URL: https://github.com/apache/kafka/pull/10342#discussion_r596445392
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java
##########
@@ -288,50 +277,28 @@ private String logPrefix() {
* Get the lock for the {@link TaskId}s directory if it is available
* @param taskId task id
* @return true if successful
- * @throws IOException if the file cannot be created or file handle cannot
be grabbed, should be considered as fatal
*/
- synchronized boolean lock(final TaskId taskId) throws IOException {
+ synchronized boolean lock(final TaskId taskId) {
if (!hasPersistentStores) {
return true;
}
- final File lockFile;
- // we already have the lock so bail out here
- final LockAndOwner lockAndOwner = locks.get(taskId);
- if (lockAndOwner != null &&
lockAndOwner.owningThread.equals(Thread.currentThread().getName())) {
- log.trace("{} Found cached state dir lock for task {}",
logPrefix(), taskId);
+ final String lockOwner = lockedTasksToStreamThreadOwner.get(taskId);
+ if (lockOwner != null) {
+ if (lockOwner.equals(Thread.currentThread().getName())) {
+ log.trace("{} Found cached state dir lock for task {}",
logPrefix(), taskId);
+ // we already own the lock
+ return true;
+ } else {
+ // another thread owns the lock
+ return false;
+ }
+ } else {
+ lockedTasksToStreamThreadOwner.put(taskId,
Thread.currentThread().getName());
+ // make sure the directory actually exists, and create it if not
+ getOrCreateDirectoryForTask(taskId);
return true;
- } else if (lockAndOwner != null) {
- // another thread owns the lock
- return false;
- }
-
- try {
- lockFile = new File(directoryForTask(taskId), LOCK_FILE_NAME);
- } catch (final ProcessorStateException e) {
- // directoryForTask could be throwing an exception if another
thread
- // has concurrently deleted the directory
Review comment:
We don't need to worry about this condition now -- another thread can
only delete the task directory if it owns the lock. In the new `lock()`
implementation, this means when we call `getOrCreateDirectroyForTask` (the new
name for this method) it's not possible for some other thread to concurrently
wipe out the task dir from beneath us, since we've already grabbed the lock by
then.
----------------------------------------------------------------
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:
[email protected]
