You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2015/09/11 20:56:16 UTC
nifi git commit: NIFI-936: Ensure that if we fail to write a record
to the repository that we do not write to that log file again
Repository: nifi
Updated Branches:
refs/heads/master 3780159d0 -> 4a7d1fe61
NIFI-936: Ensure that if we fail to write a record to the repository that we do not write to that log file again
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/4a7d1fe6
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/4a7d1fe6
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/4a7d1fe6
Branch: refs/heads/master
Commit: 4a7d1fe618b38d214c0ac11ffd2fcdca02026da6
Parents: 3780159
Author: Mark Payne <ma...@hotmail.com>
Authored: Tue Sep 8 21:16:13 2015 -0400
Committer: Mark Payne <ma...@hotmail.com>
Committed: Fri Sep 11 14:55:48 2015 -0400
----------------------------------------------------------------------
.../nifi/provenance/StandardRecordWriter.java | 334 +++++++++++--------
1 file changed, 190 insertions(+), 144 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/4a7d1fe6/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/StandardRecordWriter.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/StandardRecordWriter.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/StandardRecordWriter.java
index b95d805..ad3c6e2 100644
--- a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/StandardRecordWriter.java
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/StandardRecordWriter.java
@@ -67,21 +67,6 @@ public class StandardRecordWriter implements RecordWriter {
this.tocWriter = writer;
}
- protected void writeUUID(final DataOutputStream out, final String uuid) throws IOException {
- out.writeUTF(uuid);
- }
-
- protected void writeUUIDs(final DataOutputStream out, final Collection<String> list) throws IOException {
- if (list == null) {
- out.writeInt(0);
- } else {
- out.writeInt(list.size());
- for (final String value : list) {
- writeUUID(out, value);
- }
- }
- }
-
@Override
public synchronized File getFile() {
return file;
@@ -89,12 +74,21 @@ public class StandardRecordWriter implements RecordWriter {
@Override
public synchronized void writeHeader(final long firstEventId) throws IOException {
- lastBlockOffset = rawOutStream.getBytesWritten();
- resetWriteStream(firstEventId);
+ if (isDirty()) {
+ throw new IOException("Cannot update Provenance Repository because this Record Writer has already failed to write to the Repository");
+ }
+
+ try {
+ lastBlockOffset = rawOutStream.getBytesWritten();
+ resetWriteStream(firstEventId);
- out.writeUTF(PersistentProvenanceRepository.class.getName());
- out.writeInt(PersistentProvenanceRepository.SERIALIZATION_VERSION);
- out.flush();
+ out.writeUTF(PersistentProvenanceRepository.class.getName());
+ out.writeInt(PersistentProvenanceRepository.SERIALIZATION_VERSION);
+ out.flush();
+ } catch (final IOException ioe) {
+ markDirty();
+ throw ioe;
+ }
}
@@ -104,143 +98,173 @@ public class StandardRecordWriter implements RecordWriter {
* @throws IOException if unable to flush/close the current streams properly
*/
private void resetWriteStream(final long eventId) throws IOException {
- if ( out != null ) {
- out.flush();
- }
+ try {
+ if (out != null) {
+ out.flush();
+ }
- final long byteOffset = (byteCountingOut == null) ? rawOutStream.getBytesWritten() : byteCountingOut.getBytesWritten();
+ final long byteOffset = (byteCountingOut == null) ? rawOutStream.getBytesWritten() : byteCountingOut.getBytesWritten();
- final OutputStream writableStream;
- if ( compressed ) {
- // because of the way that GZIPOutputStream works, we need to call close() on it in order for it
- // to write its trailing bytes. But we don't want to close the underlying OutputStream, so we wrap
- // the underlying OutputStream in a NonCloseableOutputStream
- if ( out != null ) {
- out.close();
- }
+ final OutputStream writableStream;
+ if ( compressed ) {
+ // because of the way that GZIPOutputStream works, we need to call close() on it in order for it
+ // to write its trailing bytes. But we don't want to close the underlying OutputStream, so we wrap
+ // the underlying OutputStream in a NonCloseableOutputStream
+ // We don't have to check if the writer is dirty because we will have already checked before calling this method.
+ if ( out != null ) {
+ out.close();
+ }
- if ( tocWriter != null ) {
- tocWriter.addBlockOffset(rawOutStream.getBytesWritten(), eventId);
- }
+ if ( tocWriter != null ) {
+ tocWriter.addBlockOffset(rawOutStream.getBytesWritten(), eventId);
+ }
- writableStream = new BufferedOutputStream(new GZIPOutputStream(new NonCloseableOutputStream(rawOutStream), 1), 65536);
- } else {
- if ( tocWriter != null ) {
- tocWriter.addBlockOffset(rawOutStream.getBytesWritten(), eventId);
+ writableStream = new BufferedOutputStream(new GZIPOutputStream(new NonCloseableOutputStream(rawOutStream), 1), 65536);
+ } else {
+ if ( tocWriter != null ) {
+ tocWriter.addBlockOffset(rawOutStream.getBytesWritten(), eventId);
+ }
+
+ writableStream = new BufferedOutputStream(rawOutStream, 65536);
}
- writableStream = new BufferedOutputStream(rawOutStream, 65536);
+ this.byteCountingOut = new ByteCountingOutputStream(writableStream, byteOffset);
+ this.out = new DataOutputStream(byteCountingOut);
+ dirtyFlag.set(false);
+ } catch (final IOException ioe) {
+ markDirty();
+ throw ioe;
}
-
- this.byteCountingOut = new ByteCountingOutputStream(writableStream, byteOffset);
- this.out = new DataOutputStream(byteCountingOut);
}
-
@Override
public synchronized long writeRecord(final ProvenanceEventRecord record, final long recordIdentifier) throws IOException {
- final ProvenanceEventType recordType = record.getEventType();
- final long startBytes = byteCountingOut.getBytesWritten();
-
- // add a new block to the TOC if needed.
- if ( tocWriter != null && (startBytes - lastBlockOffset >= uncompressedBlockSize) ) {
- lastBlockOffset = startBytes;
+ if (isDirty()) {
+ throw new IOException("Cannot update Provenance Repository because this Record Writer has already failed to write to the Repository");
+ }
- if ( compressed ) {
- // because of the way that GZIPOutputStream works, we need to call close() on it in order for it
- // to write its trailing bytes. But we don't want to close the underlying OutputStream, so we wrap
- // the underlying OutputStream in a NonCloseableOutputStream
- resetWriteStream(recordIdentifier);
+ try {
+ final ProvenanceEventType recordType = record.getEventType();
+ final long startBytes = byteCountingOut.getBytesWritten();
+
+ // add a new block to the TOC if needed.
+ if ( tocWriter != null && (startBytes - lastBlockOffset >= uncompressedBlockSize) ) {
+ lastBlockOffset = startBytes;
+
+ if ( compressed ) {
+ // because of the way that GZIPOutputStream works, we need to call close() on it in order for it
+ // to write its trailing bytes. But we don't want to close the underlying OutputStream, so we wrap
+ // the underlying OutputStream in a NonCloseableOutputStream
+ resetWriteStream(recordIdentifier);
+ }
}
- }
- out.writeLong(recordIdentifier);
- out.writeUTF(record.getEventType().name());
- out.writeLong(record.getEventTime());
- out.writeLong(record.getFlowFileEntryDate());
- out.writeLong(record.getEventDuration());
-
- writeUUIDs(out, record.getLineageIdentifiers());
- out.writeLong(record.getLineageStartDate());
-
- writeNullableString(out, record.getComponentId());
- writeNullableString(out, record.getComponentType());
- writeUUID(out, record.getFlowFileUuid());
- writeNullableString(out, record.getDetails());
-
- // Write FlowFile attributes
- final Map<String, String> attrs = record.getPreviousAttributes();
- out.writeInt(attrs.size());
- for (final Map.Entry<String, String> entry : attrs.entrySet()) {
- writeLongString(out, entry.getKey());
- writeLongString(out, entry.getValue());
- }
+ out.writeLong(recordIdentifier);
+ out.writeUTF(record.getEventType().name());
+ out.writeLong(record.getEventTime());
+ out.writeLong(record.getFlowFileEntryDate());
+ out.writeLong(record.getEventDuration());
+
+ writeUUIDs(out, record.getLineageIdentifiers());
+ out.writeLong(record.getLineageStartDate());
+
+ writeNullableString(out, record.getComponentId());
+ writeNullableString(out, record.getComponentType());
+ writeUUID(out, record.getFlowFileUuid());
+ writeNullableString(out, record.getDetails());
+
+ // Write FlowFile attributes
+ final Map<String, String> attrs = record.getPreviousAttributes();
+ out.writeInt(attrs.size());
+ for (final Map.Entry<String, String> entry : attrs.entrySet()) {
+ writeLongString(out, entry.getKey());
+ writeLongString(out, entry.getValue());
+ }
- final Map<String, String> attrUpdates = record.getUpdatedAttributes();
- out.writeInt(attrUpdates.size());
- for (final Map.Entry<String, String> entry : attrUpdates.entrySet()) {
- writeLongString(out, entry.getKey());
- writeLongNullableString(out, entry.getValue());
- }
+ final Map<String, String> attrUpdates = record.getUpdatedAttributes();
+ out.writeInt(attrUpdates.size());
+ for (final Map.Entry<String, String> entry : attrUpdates.entrySet()) {
+ writeLongString(out, entry.getKey());
+ writeLongNullableString(out, entry.getValue());
+ }
- // If Content Claim Info is present, write out a 'TRUE' followed by claim info. Else, write out 'false'.
- if (record.getContentClaimSection() != null && record.getContentClaimContainer() != null && record.getContentClaimIdentifier() != null) {
- out.writeBoolean(true);
- out.writeUTF(record.getContentClaimContainer());
- out.writeUTF(record.getContentClaimSection());
- out.writeUTF(record.getContentClaimIdentifier());
- if (record.getContentClaimOffset() == null) {
- out.writeLong(0L);
+ // If Content Claim Info is present, write out a 'TRUE' followed by claim info. Else, write out 'false'.
+ if (record.getContentClaimSection() != null && record.getContentClaimContainer() != null && record.getContentClaimIdentifier() != null) {
+ out.writeBoolean(true);
+ out.writeUTF(record.getContentClaimContainer());
+ out.writeUTF(record.getContentClaimSection());
+ out.writeUTF(record.getContentClaimIdentifier());
+ if (record.getContentClaimOffset() == null) {
+ out.writeLong(0L);
+ } else {
+ out.writeLong(record.getContentClaimOffset());
+ }
+ out.writeLong(record.getFileSize());
} else {
- out.writeLong(record.getContentClaimOffset());
+ out.writeBoolean(false);
}
- out.writeLong(record.getFileSize());
- } else {
- out.writeBoolean(false);
- }
- // If Previous Content Claim Info is present, write out a 'TRUE' followed by claim info. Else, write out 'false'.
- if (record.getPreviousContentClaimSection() != null && record.getPreviousContentClaimContainer() != null && record.getPreviousContentClaimIdentifier() != null) {
- out.writeBoolean(true);
- out.writeUTF(record.getPreviousContentClaimContainer());
- out.writeUTF(record.getPreviousContentClaimSection());
- out.writeUTF(record.getPreviousContentClaimIdentifier());
- if (record.getPreviousContentClaimOffset() == null) {
- out.writeLong(0L);
+ // If Previous Content Claim Info is present, write out a 'TRUE' followed by claim info. Else, write out 'false'.
+ if (record.getPreviousContentClaimSection() != null && record.getPreviousContentClaimContainer() != null && record.getPreviousContentClaimIdentifier() != null) {
+ out.writeBoolean(true);
+ out.writeUTF(record.getPreviousContentClaimContainer());
+ out.writeUTF(record.getPreviousContentClaimSection());
+ out.writeUTF(record.getPreviousContentClaimIdentifier());
+ if (record.getPreviousContentClaimOffset() == null) {
+ out.writeLong(0L);
+ } else {
+ out.writeLong(record.getPreviousContentClaimOffset());
+ }
+
+ if (record.getPreviousFileSize() == null) {
+ out.writeLong(0L);
+ } else {
+ out.writeLong(record.getPreviousFileSize());
+ }
} else {
- out.writeLong(record.getPreviousContentClaimOffset());
+ out.writeBoolean(false);
}
- if (record.getPreviousFileSize() == null) {
- out.writeLong(0L);
- } else {
- out.writeLong(record.getPreviousFileSize());
+ // write out the identifier of the destination queue.
+ writeNullableString(out, record.getSourceQueueIdentifier());
+
+ // Write type-specific info
+ if (recordType == ProvenanceEventType.FORK || recordType == ProvenanceEventType.JOIN || recordType == ProvenanceEventType.CLONE || recordType == ProvenanceEventType.REPLAY) {
+ writeUUIDs(out, record.getParentUuids());
+ writeUUIDs(out, record.getChildUuids());
+ } else if (recordType == ProvenanceEventType.RECEIVE) {
+ writeNullableString(out, record.getTransitUri());
+ writeNullableString(out, record.getSourceSystemFlowFileIdentifier());
+ } else if (recordType == ProvenanceEventType.SEND) {
+ writeNullableString(out, record.getTransitUri());
+ } else if (recordType == ProvenanceEventType.ADDINFO) {
+ writeNullableString(out, record.getAlternateIdentifierUri());
+ } else if (recordType == ProvenanceEventType.ROUTE) {
+ writeNullableString(out, record.getRelationship());
}
- } else {
- out.writeBoolean(false);
- }
- // write out the identifier of the destination queue.
- writeNullableString(out, record.getSourceQueueIdentifier());
-
- // Write type-specific info
- if (recordType == ProvenanceEventType.FORK || recordType == ProvenanceEventType.JOIN || recordType == ProvenanceEventType.CLONE || recordType == ProvenanceEventType.REPLAY) {
- writeUUIDs(out, record.getParentUuids());
- writeUUIDs(out, record.getChildUuids());
- } else if (recordType == ProvenanceEventType.RECEIVE) {
- writeNullableString(out, record.getTransitUri());
- writeNullableString(out, record.getSourceSystemFlowFileIdentifier());
- } else if (recordType == ProvenanceEventType.SEND) {
- writeNullableString(out, record.getTransitUri());
- } else if (recordType == ProvenanceEventType.ADDINFO) {
- writeNullableString(out, record.getAlternateIdentifierUri());
- } else if (recordType == ProvenanceEventType.ROUTE) {
- writeNullableString(out, record.getRelationship());
+ out.flush();
+ recordCount++;
+ return byteCountingOut.getBytesWritten() - startBytes;
+ } catch (final IOException ioe) {
+ markDirty();
+ throw ioe;
}
+ }
- out.flush();
- recordCount++;
- return byteCountingOut.getBytesWritten() - startBytes;
+ protected void writeUUID(final DataOutputStream out, final String uuid) throws IOException {
+ out.writeUTF(uuid);
+ }
+
+ protected void writeUUIDs(final DataOutputStream out, final Collection<String> list) throws IOException {
+ if (list == null) {
+ out.writeInt(0);
+ } else {
+ out.writeInt(list.size());
+ for (final String value : list) {
+ writeUUID(out, value);
+ }
+ }
}
protected void writeNullableString(final DataOutputStream out, final String toWrite) throws IOException {
@@ -274,20 +298,33 @@ public class StandardRecordWriter implements RecordWriter {
lock();
try {
try {
- if (out != null) {
- try {
- out.flush();
- } finally {
- out.close();
- }
+ // We want to close 'out' only if the writer is not 'dirty'.
+ // If the writer is dirty, then there was a failure to write
+ // to disk, which means that we likely have a partial record written
+ // to disk.
+ //
+ // If we call close() on out, it will in turn call flush() on the underlying
+ // output stream, which is a BufferedOutputStream. As a result, we will end
+ // up flushing the buffer after a partially written record, which results in
+ // essentially random bytes being written to the repository, which causes
+ // corruption and un-recoverability. Since we will close the underlying 'rawOutStream'
+ // below, we will still appropriately clean up the resources help by this writer, so
+ // we are still OK in terms of closing all resources held by the writer.
+ if (out != null && !isDirty()) {
+ out.close();
}
} finally {
- rawOutStream.close();
-
- if ( tocWriter != null ) {
- tocWriter.close();
+ try {
+ rawOutStream.close();
+ } finally {
+ if (tocWriter != null) {
+ tocWriter.close();
+ }
}
}
+ } catch (final IOException ioe) {
+ markDirty();
+ throw ioe;
} finally {
unlock();
}
@@ -330,10 +367,15 @@ public class StandardRecordWriter implements RecordWriter {
@Override
public void sync() throws IOException {
- if ( tocWriter != null ) {
- tocWriter.sync();
+ try {
+ if ( tocWriter != null ) {
+ tocWriter.sync();
+ }
+ fos.getFD().sync();
+ } catch (final IOException ioe) {
+ markDirty();
+ throw ioe;
}
- fos.getFD().sync();
}
@Override
@@ -345,4 +387,8 @@ public class StandardRecordWriter implements RecordWriter {
public void markDirty() {
dirtyFlag.set(true);
}
+
+ public boolean isDirty() {
+ return dirtyFlag.get();
+ }
}