Fix the data loss issue caused by the cache expiration in PartitionerDataWriter
Fix the data loss issue caused by the cache
expiration in PartitionerDataWriter
address comments
address comments and avoid overflow
Closes from ZihanLi58/GOBBLIN-1343
@@ -101,6 +101,7 @@ public class PartitionedDataWriter<S, D> extends WriterWrapper<D> implements Fin
private final ControlMessageHandler controlMessageHandler;
private boolean isSpeculativeAttemptSafe;
private boolean isWatermarkCapable;
+ private long writeTimeoutInterval;
private ScheduledExecutorService cacheCleanUpExecutor;
@@ -128,6 +129,7 @@ public class PartitionedDataWriter<S, D> extends WriterWrapper<D> implements Fin
this.state.setProp(WRITER_LATEST_SCHEMA, builder.getSchema());
}
long cacheExpiryInterval = this.state.getPropAsLong(PARTITIONED_WRITER_CACHE_TTL_SECONDS, DEFAULT_PARTITIONED_WRITER_CACHE_TTL_SECONDS);
+ this.writeTimeoutInterval = cacheExpiryInterval / 3;
log.debug("PartitionedDataWriter: Setting cache expiry interval to {} seconds", cacheExpiryInterval);
this.partitionWriters = CacheBuilder.newBuilder()
@@ -235,18 +237,24 @@ public class PartitionedDataWriter<S, D> extends WriterWrapper<D> implements Fin
@Override
public void writeEnvelope(RecordEnvelope<D> recordEnvelope) throws IOException {
try {
- DataWriter<D> writer = getDataWriterForRecord(recordEnvelope.getRecord());
+ GenericRecord partition = getPartitionForRecord(recordEnvelope.getRecord());
+ DataWriter<D> writer = this.partitionWriters.get(partition);
+ long startTime = System.currentTimeMillis();
writer.writeEnvelope(recordEnvelope);
+ long timeForWriting = System.currentTimeMillis() - startTime;
+ // If the write take a long time, which is 1/3 of cache expiration time, we fail the writer to avoid data loss
+ // and further slowness on the same HDFS block
+ if (timeForWriting / 1000 > this.writeTimeoutInterval ) {
+ throw new IOException(String.format("Write record took %s s, but threshold is %s s",
+ timeForWriting / 1000, writeTimeoutInterval));
+ }
} catch (ExecutionException ee) {
throw new IOException(ee);
}
}
- private DataWriter<D> getDataWriterForRecord(D record)
- throws ExecutionException {
- GenericRecord partition =
- this.shouldPartition ? this.partitioner.get().partitionForRecord(record) : NON_PARTITIONED_WRITER_KEY;
- return this.partitionWriters.get(partition);
+ private GenericRecord getPartitionForRecord(D record) {
+ return this.shouldPartition ? this.partitioner.get().partitionForRecord(record) : NON_PARTITIONED_WRITER_KEY;
}
@Override
