swathimocharla19
commented
5 years ago To overcome this issue, the following change was made in FileUtils.java
public static FileStatus fileStatusWithMaxOffset(
Storage storage,
Path path,
CommittedFileFilter filter
) {
if (!storage.exists(path.toString())) {
return null;
}
long maxOffset = -1L;
FileStatus fileStatusWithMaxOffset = null;
List <FileStatus> statuses = storage.list(path.toString());
for (FileStatus status: statuses) {
if (status.isDirectory()) {
FileStatus fileStatus = fileStatusWithMaxOffset(storage, status.getPath(), filter);
if (fileStatus != null) {
if (fileStatus.getPath().getName().contains(".parquet") && fileStatus.getLen() < 8) {
log.trace("Found an empty parquet file and deleting it: {}", fileStatus.getPath().getName());
storage.delete(fileStatus.getPath().toString());
} else {
long offset = extractOffset(fileStatus.getPath().getName());
if (offset > maxOffset) {
maxOffset = offset;
fileStatusWithMaxOffset = fileStatus;
}
}
}
} else {
String filename = status.getPath().getName();
log.trace("Checked for max offset: {}", status.getPath());
if (filter.accept(status.getPath())) {
if (status.getPath().getName().contains(".parquet") && status.getLen() < 8) {
log.trace("Found an empty parquet file and deleting it: {}", status.getPath().getName());
storage.delete(status.getPath().toString());
} else {
long offset = extractOffset(filename);
if (offset > maxOffset) {
maxOffset = offset;
fileStatusWithMaxOffset = status;
}
}
}
}
}
return fileStatusWithMaxOffset;
}To test this, after inserting data into HDFS, the latest file (with max offsets) was deleted and re-created , the size is now 0. At this point the connect worker was restarted, and as part of the restart, the 0 sized parquet file is deleted. Now, the previous parquet file becomes the latest and the offsets are read from there. So, the deleted offsets parquet file is recreated, given that Kafka still holds this data at this given point.
ncliang
RossierFl
yifatshani
shmyer
We've run into an issue where sometimes when Kafka HDFS connect is stopped for any reason, an empty parquet file is generated in HDFS. After the connect worker is started, the subsequent writes fail on HDFS as the worker tries to gather the schema from the latest available parquet file, based on its offsets (which is empty). This situation occurs when "schema.compatibility": "BACKWARD"