Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

File locked during deletion - fix #14

Merged
merged 6 commits into from
Sep 29, 2023
Merged

File locked during deletion - fix #14

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b3b58f5
add filelock before deletion
apoorva918 Sep 14, 2023
e3f8d8d
updated version
apoorva918 Sep 19, 2023
dc38aa1
fix last file deletion delay
apoorva918 Sep 20, 2023
1d706ae
add filelock to parquet ingest service
apoorva918 Sep 25, 2023
da5f0a2
formatting change
apoorva918 Sep 25, 2023
b7637f5
fixed log messages
apoorva918 Sep 28, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 52 additions & 35 deletions ...sor-collector/src/main/java/io/pravega/sensor/collector/parquet/ParquetFileProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,13 @@
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.sql.Connection;
import java.util.ArrayList;
import java.util.Collections;
Expand Down Expand Up @@ -92,11 +95,12 @@ public static ParquetFileProcessor create(ParquetFileConfig config, EventStreamC

public void ingestParquetFiles() throws Exception {
log.trace("ingestParquetFiles: BEGIN");
findAndRecordNewFiles();
processNewFiles();
// delete leftover completed files
if (config.enableDeleteCompletedFiles) {
deleteCompletedFiles();
}
findAndRecordNewFiles();
processNewFiles();
log.trace("ingestParquetFiles: END");
}

Expand Down Expand Up @@ -185,33 +189,35 @@ void processFile(FileNameWithOffset fileNameWithBeginOffset, long firstSequenceN
writer.abort();

try (final InputStream inputStream = new FileInputStream(fileNameWithBeginOffset.fileName)) {
final CountingInputStream countingInputStream = new CountingInputStream(inputStream);
countingInputStream.skip(fileNameWithBeginOffset.offset);
final Pair<Long,Long> result = eventGenerator.generateEventsFromInputStream(countingInputStream, firstSequenceNumber,
e -> {
log.trace("processFile: event={}", e);
try {
writer.writeEvent(e.routingKey, e.bytes);
numofbytes.addAndGet(e.bytes.length);
} catch (TxnFailedException ex) {
throw new RuntimeException(ex);
}
});
final Optional<UUID> txnId = writer.flush();
final long nextSequenceNumber = result.getLeft();
final long endOffset = result.getRight();
state.addCompletedFile(fileNameWithBeginOffset.fileName, fileNameWithBeginOffset.offset, endOffset, nextSequenceNumber, txnId);
// injectCommitFailure();
writer.commit();
state.deleteTransactionToCommit(txnId);
try(final CountingInputStream countingInputStream = new CountingInputStream(inputStream)) {
countingInputStream.skip(fileNameWithBeginOffset.offset);
final Pair<Long,Long> result = eventGenerator.generateEventsFromInputStream(countingInputStream, firstSequenceNumber,
e -> {
log.trace("processFile: event={}", e);
try {
writer.writeEvent(e.routingKey, e.bytes);
numofbytes.addAndGet(e.bytes.length);

} catch (TxnFailedException ex) {
throw new RuntimeException(ex);
}
});
final Optional<UUID> txnId = writer.flush();
final long nextSequenceNumber = result.getLeft();
final long endOffset = result.getRight();
state.addCompletedFile(fileNameWithBeginOffset.fileName, fileNameWithBeginOffset.offset, endOffset, nextSequenceNumber, txnId);
// injectCommitFailure();
writer.commit();
state.deleteTransactionToCommit(txnId);

double elapsedSec = (System.nanoTime() - timestamp) / 1_000_000_000.0;
double megabyteCount = numofbytes.getAndSet(0) / 1_000_000.0;
double megabytesPerSec = megabyteCount / elapsedSec;
log.info("processFile: Finished ingesting file {}; endOffset={}, nextSequenceNumber={}",
fileNameWithBeginOffset.fileName, endOffset, nextSequenceNumber);
log.info("Sent {} MB in {} sec", megabyteCount, elapsedSec );
log.info("Transfer rate: {} MB/sec", megabytesPerSec);
double elapsedSec = (System.nanoTime() - timestamp) / 1_000_000_000.0;
double megabyteCount = numofbytes.getAndSet(0) / 1_000_000.0;
double megabytesPerSec = megabyteCount / elapsedSec;
log.info("processFile: Finished ingesting file {}; endOffset={}, nextSequenceNumber={}",
fileNameWithBeginOffset.fileName, endOffset, nextSequenceNumber);
log.info("Sent {} MB in {} sec", megabyteCount, elapsedSec );
log.info("Transfer rate: {} MB/sec", megabytesPerSec);
}
}

// Delete file right after ingesting
Expand All @@ -224,14 +230,25 @@ void processFile(FileNameWithOffset fileNameWithBeginOffset, long firstSequenceN
void deleteCompletedFiles() throws Exception {
final List<FileNameWithOffset> completedFiles = state.getCompletedFiles();
completedFiles.forEach(file -> {
try {
Files.deleteIfExists(Paths.get(file.fileName));
log.info("deleteCompletedFiles: Deleted file {}", file.fileName);
// Only remove from database if we could delete file.
state.deleteCompletedFile(file.fileName);
//Obtain a lock on file
try(FileChannel channel = FileChannel.open(Paths.get(file.fileName),StandardOpenOption.WRITE)){
try(FileLock lock = channel.tryLock()) {
if(lock!=null){
Files.deleteIfExists(Paths.get(file.fileName));
log.info("deleteCompletedFiles: Deleted file {}", file.fileName);
lock.release();
// Only remove from database if we could delete file.
state.deleteCompletedFile(file.fileName);
}
else{
log.warn("Unable to obtain lock on file {}. File is locked by another process.", file.fileName);
throw new Exception();
}
}
} catch (Exception e) {
log.warn("Unable to delete ingested file {}", e);
// We can continue on this error. It will be retried on the next iteration.
log.warn("Unable to delete ingested file {}", e.getMessage());
log.warn("Deletion will be retried on the next iteration.");
// We can continue on this error. Deletion will be retried on the next iteration.
}
});
}
Expand Down
2 changes: 1 addition & 1 deletion ...ga-sensor-collector/src/main/java/io/pravega/sensor/collector/rawfile/EventGenerator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,6 @@ protected Pair<Long, Long> generateEventsFromInputStream(CountingInputStream inp
} catch (Exception e){
log.error("Exception = {}",e);
throw e;
}
}
}
}
88 changes: 52 additions & 36 deletions ...-sensor-collector/src/main/java/io/pravega/sensor/collector/rawfile/RawFileProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,13 @@
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.sql.Connection;
import java.util.ArrayList;
import java.util.Collections;
Expand Down Expand Up @@ -91,11 +94,12 @@ public static RawFileProcessor create(RawFileConfig config, EventStreamClientFac

public void ingestRawFiles() throws Exception {
log.trace("ingestRawFiles: BEGIN");
findAndRecordNewFiles();
processNewFiles();
// delete leftover completed files
if (config.enableDeleteCompletedFiles) {
deleteCompletedFiles();
}
findAndRecordNewFiles();
processNewFiles();
log.trace("ingestRawFiles: END");
}

Expand Down Expand Up @@ -182,34 +186,35 @@ void processFile(FileNameWithOffset fileNameWithBeginOffset, long firstSequenceN
writer.abort();

try (final InputStream inputStream = new FileInputStream(fileNameWithBeginOffset.fileName)) {
final CountingInputStream countingInputStream = new CountingInputStream(inputStream);
countingInputStream.skip(fileNameWithBeginOffset.offset);
final Pair<Long,Long> result = eventGenerator.generateEventsFromInputStream(countingInputStream, firstSequenceNumber,
e -> {
log.trace("processFile: event={}", e);
try {
writer.writeEvent(e.routingKey, e.bytes);
numofbytes.addAndGet(e.bytes.length);

} catch (TxnFailedException ex) {
throw new RuntimeException(ex);
}
});
final Optional<UUID> txnId = writer.flush();
final long nextSequenceNumber = result.getLeft();
final long endOffset = result.getRight();
state.addCompletedFile(fileNameWithBeginOffset.fileName, fileNameWithBeginOffset.offset, endOffset, nextSequenceNumber, txnId);
// injectCommitFailure();
writer.commit();
state.deleteTransactionToCommit(txnId);
try(final CountingInputStream countingInputStream = new CountingInputStream(inputStream)) {
countingInputStream.skip(fileNameWithBeginOffset.offset);
final Pair<Long,Long> result = eventGenerator.generateEventsFromInputStream(countingInputStream, firstSequenceNumber,
e -> {
log.trace("processFile: event={}", e);
try {
writer.writeEvent(e.routingKey, e.bytes);
numofbytes.addAndGet(e.bytes.length);

} catch (TxnFailedException ex) {
throw new RuntimeException(ex);
}
});
final Optional<UUID> txnId = writer.flush();
final long nextSequenceNumber = result.getLeft();
final long endOffset = result.getRight();
state.addCompletedFile(fileNameWithBeginOffset.fileName, fileNameWithBeginOffset.offset, endOffset, nextSequenceNumber, txnId);
// injectCommitFailure();
writer.commit();
state.deleteTransactionToCommit(txnId);

double elapsedSec = (System.nanoTime() - timestamp) / 1_000_000_000.0;
double megabyteCount = numofbytes.getAndSet(0) / 1_000_000.0;
double megabytesPerSec = megabyteCount / elapsedSec;
log.info("processFile: Finished ingesting file {}; endOffset={}, nextSequenceNumber={}",
fileNameWithBeginOffset.fileName, endOffset, nextSequenceNumber);
log.info("Sent {} MB in {} sec", megabyteCount, elapsedSec );
log.info("Transfer rate: {} MB/sec", megabytesPerSec);
double elapsedSec = (System.nanoTime() - timestamp) / 1_000_000_000.0;
double megabyteCount = numofbytes.getAndSet(0) / 1_000_000.0;
double megabytesPerSec = megabyteCount / elapsedSec;
log.info("processFile: Finished ingesting file {}; endOffset={}, nextSequenceNumber={}",
fileNameWithBeginOffset.fileName, endOffset, nextSequenceNumber);
log.info("Sent {} MB in {} sec", megabyteCount, elapsedSec );
log.info("Transfer rate: {} MB/sec", megabytesPerSec);
}
}

// Delete file right after ingesting
Expand All @@ -221,14 +226,25 @@ void processFile(FileNameWithOffset fileNameWithBeginOffset, long firstSequenceN
void deleteCompletedFiles() throws Exception {
final List<FileNameWithOffset> completedFiles = state.getCompletedFiles();
completedFiles.forEach(file -> {
try {
Files.deleteIfExists(Paths.get(file.fileName));
log.info("deleteCompletedFiles: Deleted file {}", file.fileName);
// Only remove from database if we could delete file.
state.deleteCompletedFile(file.fileName);
//Obtain a lock on file
try(FileChannel channel = FileChannel.open(Paths.get(file.fileName),StandardOpenOption.WRITE)){
try(FileLock lock = channel.tryLock()) {
if(lock!=null){
Files.deleteIfExists(Paths.get(file.fileName));
log.info("deleteCompletedFiles: Deleted file {}", file.fileName);
lock.release();
// Only remove from database if we could delete file.
state.deleteCompletedFile(file.fileName);
}
else{
log.warn("Unable to obtain lock on file {}. File is locked by another process.", file.fileName);
throw new Exception();
}
}
} catch (Exception e) {
log.warn("Unable to delete ingested file {}", e);
// We can continue on this error. It will be retried on the next iteration.
log.warn("Unable to delete ingested file {}", e.getMessage());
log.warn("Deletion will be retried on the next iteration.");
// We can continue on this error. Deletion will be retried on the next iteration.
}
});
}
Expand Down
2 changes: 1 addition & 1 deletion scripts/env.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,5 +11,5 @@ export ENV_LOCAL_SCRIPT=$(dirname $0)/env-local.sh
if [[ -f ${ENV_LOCAL_SCRIPT} ]]; then
source ${ENV_LOCAL_SCRIPT}
fi
export APP_VERSION=${APP_VERSION:-0.2.16}
export APP_VERSION=${APP_VERSION:-0.2.17}
export GRADLE_OPTIONS="${GRADLE_OPTIONS:-"-Pversion=${APP_VERSION}"}"
Loading