21

I have a piece of code that monitors a directory for addition of files. Whenever a new file is added to the directory, the contents of the file are picked and published on kafka and then the file is deleted.

This works when I make a single request but as soon as I subject my code to 5 or 10 user request from jMeter, the contents are published on kafka successfully but the code isn't able to delete the file. I get a FileSystemException with a message that The process cannot access the file because it is being used by another process..

I guess there is some concurrency issue which I am unable to see.

public void monitor() throws IOException, InterruptedException {
    Path faxFolder = Paths.get(TEMP_FILE_LOCATION);
    WatchService watchService = FileSystems.getDefault().newWatchService();
    faxFolder.register(watchService, StandardWatchEventKinds.ENTRY_CREATE);
    boolean valid = true;
    do {
        WatchKey watchKey = watchService.take();
        for (WatchEvent<?> event : watchKey.pollEvents()) {
            if (StandardWatchEventKinds.ENTRY_CREATE.equals(event.kind())) {
                String fileName = event.context().toString();
                publishToKafka(new File(TEMP_FILE_LOCATION + fileName).toPath(), "topic");
            }
        }
        valid = watchKey.reset();
    } while (valid);
}

private void publishToKafka(Path path, String topic) {
    try (BufferedReader reader = Files.newBufferedReader(path)) {
        String input = null;
        while ((input = reader.readLine()) != null) {
            kafkaProducer.publishMessageOnTopic(input, topic);
        }
    } catch (IOException e) {
        LOG.error("Could not read buffered file to send message on kafka.", e);
    } finally {
        try {
            Files.deleteIfExists(path); // This is where I get the exception
        } catch (IOException e) {
            LOG.error("Problem in deleting the buffered file {}.", path.getFileName(), e);
        }
    }
}

Exception Log : 

java.nio.file.FileSystemException: D:\upload\notif-1479974962595.csv: The process cannot access the file because it is being used by another process.

    at sun.nio.fs.WindowsException.translateToIOException(Unknown Source)
    at sun.nio.fs.WindowsException.rethrowAsIOException(Unknown Source)
    at sun.nio.fs.WindowsException.rethrowAsIOException(Unknown Source)
    at sun.nio.fs.WindowsFileSystemProvider.implDelete(Unknown Source)
    at sun.nio.fs.AbstractFileSystemProvider.deleteIfExists(Unknown Source)
    at java.nio.file.Files.deleteIfExists(Unknown Source)
    at com.panasonic.mdw.core.utils.MonitorDirectory$FileContentPublisher.publishToKafka(MonitorDirectory.java:193)
    at com.panasonic.mdw.core.utils.MonitorDirectory$FileContentPublisher.sendData(MonitorDirectory.java:125)
    at com.panasonic.mdw.core.utils.MonitorDirectory$FileContentPublisher.run(MonitorDirectory.java:113)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
    at java.lang.Thread.run(Unknown Source)
Erik
  • 503
  • 1
  • 7
  • 26
sensitive_piece_of_horseflesh
  • 909
  • 4
  • 16
  • 40
  • 2
    Could it be that the process that creates the file hasn't finished creating it (hasn't closed it) before you try to read it and delete it? – Klitos Kyriacou Nov 24 '16 at 10:50
  • 1
    By the way, the WatchService is closeable, so put `newWatchService()` inside a try-with-resources. Also, since you're using nio everywhere else, for consistency instead of `new File(...).toPath()` do `faxFolder.resolve(filename)`. – Klitos Kyriacou Nov 24 '16 at 11:03
  • 3
    @Klitos Kyriacou: since the `WatchEvent`’s context is already a `Path` when watching a file system, even the conversion to string is obsolete and `faxFolder.resolve((Path)event.context())` is sufficient. I think, you’re right, reacting on the creation event immediately doesn’t give the creator enough time to close the file. Interestingly, on my system, that has the opposite effect: reading the file fails as it is still be written, but deleting works, as the system postpones the actual removal to the point, the creator closes the file. – Holger Nov 24 '16 at 11:20
  • @Holger good points. I think (need to test) that when a new file is created, you get two events: first, the entry creation, followed by an entry modification (when the datestamp is updated on closing the file). The file should only be read after the modification event. (And maybe after a timeout, in case an empty file is created and left there.) – Klitos Kyriacou Nov 24 '16 at 11:53
  • 1
    @Klitos Kyriacou: there can be more than one modification event while the other app is writing into the file. So you have to setup a timer that initiates your action a certain time after the last modification event, resetting the timer on each modification. And that’s still only handling the case that the other app creates the file, writes to it and closes it straight-forwardly… – Holger Nov 24 '16 at 13:26
  • @KlitosKyriacou and @Holger you are correct , this was exactly the problem, the event was file creation , so while the file was still being written by another thread, the `publishToKafka()` method would find `null` in between and consider that an end of input and then the code goes ahead for deletion which is not possible right now. – sensitive_piece_of_horseflesh Nov 24 '16 at 16:27
  • i thought of this on my way back home :P and have tried a different approach now which does not involve using a directory watcher. – sensitive_piece_of_horseflesh Nov 24 '16 at 16:29

4 Answers4

3

Looking at your code it seems when one file is picked by thread for publishing again another thread is picking it up for publishing. That's why no one is able to delete it. It must be concurrency issue only. You should redesign code based up on criterion : steps which can be run concurrently and those which cannot be. So steps in the entire process are :

  1. pick up a file (main thread should do it)
  2. publish a file (call other thread to do it )
  3. delete the file (called thread should delete it)
  4. check if any file present (again main thread can do it)

Also the moment a file is selected, you can read it into buffer , delete it and then continue with publish. This will make sure that main thread does not assign this file to some other thread.

Goro
  • 516
  • 4
  • 15
2

It is always a better idea to add sleep time in WatchService events:

if (StandardWatchEventKinds.ENTRY_CREATE.equals(event.kind())) {

       try {
            Thread.sleep(3000);
       } catch (InterruptedException e) {
            e.printStackTrace();
       }

// now do your intended jobs ...

I needed to add sleep time otherwise it wouldn't work for multiple requests and used to get the error:

The process cannot access the file because it is being used by another process
BSMP
  • 4,596
  • 8
  • 33
  • 44
Krishna Kothoor
  • 41
  • 4
1

I had similar problem as well as in the following thread: Multithreading on Queue when trying to upload a dynamically created files to a Queue service and it was taking 2 days for me to resolve. Thanks to Holger who gave answer as above whereby the locking occurs might be due to the creation had not fully done upon reading by another thread, it has saved me a lot of time.

My initial solution as found a lot from the internet, was:

WatchEvent<Path> ev = cast(event);
Path name = ev.context();
Path child = dir.resolve(name);
//queueUploadFile(child);
if (kind == ENTRY_CREATE) {
    uploadToQueue(this.queueId, child);
}

I changed it to: 

WatchEvent<Path> ev = cast(event);
Path name = ev.context();
Path child = dir.resolve(name);
//queueUploadFile(child);
if (kind == ENTRY_MODIFY) {
    uploadToQueue(this.queueId, child);
}

And everything works perfectly. To handle the "somehow" multiple ENTRY_MODIFY events firing (duplicated files uploaded), I perform deletion on the file inside the uploadToQueue() method once it's uploaded.

I hope my approach taken based on the above contribution will also help others with similar problem.

I Made Putrama
  • 11
  • 3
0

You have to close all the connection which is accessing that file before deleting it.

Arun Kasyakar
  • 905
  • 8
  • 9