public void pollNodesAndCopyFile(String dir, String filter, String destDir) {
		int nodeNo = 0;
		List<String> tmpFiles = null;
		String responseTopic = null;

		Object lock = new Object();
		boolean errors = false;
		ExecutorService taskExecutor = Executors
				.newFixedThreadPool(NUM_FILE_COPY_THREADS);

		String dataDir = (configs.getLocalDataDir() + "/" + dir).replace("//",
				"/");
		try {
			String filePath;
			String refId;
			for (String node : nodes) {
				refId = uuidGen.generateTimeBasedUUID().toString();
				responseTopic = TwisterConstants.PARTITION_FILE_RESPONSE_TOPIC_BASE
						+ refId;
				DirListRequest req = new DirListRequest(dataDir, filter,
						responseTopic);
				req.setRefMessageId(refId);

				List<TwisterMessage> responses = call.sendReceive(
						TwisterConstants.MAP_REDUCE_TOPIC_BASE + "/" + nodeNo,
						responseTopic, req);
				if (responses == null) {
					System.err.print("Node " + node + " is not responding");
				} else if (responses.get(0).readByte() == TwisterConstants.WORKER_RESPONSE) {
					WorkerResponse res = new WorkerResponse(responses.get(0));			
					if (res.isHasException()) {
						System.err.println("Node " + node + " gave error: "
								+ res.getExceptionString());
						continue;
					}
				} else {
					DirListResponse response = new DirListResponse();
					response.fromTwisterMessage(responses.get(0));
					tmpFiles = response.getFileNames();
					if (tmpFiles != null) {
						for (String file : tmpFiles) {
							filePath = node + ":" + file;
							taskExecutor.execute(new FileCopyThread(filePath,
									destDir, lock, errors));
						}
					}
				}
				nodeNo++;
			}

			taskExecutor.shutdown();
			try {
				taskExecutor.awaitTermination(Long.MAX_VALUE,
						TimeUnit.NANOSECONDS);
			} catch (InterruptedException e) {
				System.exit(-1);
			}

			if (errors) {
				System.exit(-1);
			} else {
				System.exit(0);
			}

		} catch (Exception e) {
			System.err
					.println("File collector failed due to " + e.getMessage());
			System.exit(-1);
		}
	}