You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@manifoldcf.apache.org by mo...@apache.org on 2014/10/19 00:26:21 UTC
svn commit: r1632854 -
/manifoldcf/trunk/connectors/hdfs/connector/src/main/java/org/apache/manifoldcf/agents/output/hdfs/HDFSOutputConnector.java
Author: molgun
Date: Sat Oct 18 22:26:20 2014
New Revision: 1632854
URL: http://svn.apache.org/r1632854
Log:
Added activity logging for HDFS output connector.
Modified:
manifoldcf/trunk/connectors/hdfs/connector/src/main/java/org/apache/manifoldcf/agents/output/hdfs/HDFSOutputConnector.java
Modified: manifoldcf/trunk/connectors/hdfs/connector/src/main/java/org/apache/manifoldcf/agents/output/hdfs/HDFSOutputConnector.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/connectors/hdfs/connector/src/main/java/org/apache/manifoldcf/agents/output/hdfs/HDFSOutputConnector.java?rev=1632854&r1=1632853&r2=1632854&view=diff
==============================================================================
--- manifoldcf/trunk/connectors/hdfs/connector/src/main/java/org/apache/manifoldcf/agents/output/hdfs/HDFSOutputConnector.java (original)
+++ manifoldcf/trunk/connectors/hdfs/connector/src/main/java/org/apache/manifoldcf/agents/output/hdfs/HDFSOutputConnector.java Sat Oct 18 22:26:20 2014
@@ -286,10 +286,11 @@ public class HDFSOutputConnector extends
Path path = new Path(strBuff.toString());
Long startTime = new Long(System.currentTimeMillis());
- createFile(path, document.getBinaryStream());
+ createFile(path, document.getBinaryStream(),activities,documentURI);
activities.recordActivity(startTime, INGEST_ACTIVITY, new Long(document.getBinaryLength()), documentURI, "OK", null);
return DOCUMENTSTATUS_ACCEPTED;
} catch (URISyntaxException e) {
+ activities.recordActivity(null,INGEST_ACTIVITY,new Long(document.getBinaryLength()),documentURI,activities.EXCEPTION,"Rejected due to URISyntaxException");
handleURISyntaxException(e);
return DOCUMENTSTATUS_REJECTED;
}
@@ -320,11 +321,13 @@ public class HDFSOutputConnector extends
strBuff.append(documentURItoFilePath(documentURI));
Path path = new Path(strBuff.toString());
Long startTime = new Long(System.currentTimeMillis());
- deleteFile(path);
+ deleteFile(path,activities,documentURI);
activities.recordActivity(startTime, REMOVE_ACTIVITY, null, documentURI, "OK", null);
} catch (JSONException e) {
+ activities.recordActivity(null,REMOVE_ACTIVITY,null,documentURI,activities.EXCEPTION,"Rejected due to JSONException.");
handleJSONException(e);
} catch (URISyntaxException e) {
+ activities.recordActivity(null,REMOVE_ACTIVITY,null,documentURI,activities.EXCEPTION,"Rejected due to URISyntaxException.");
handleURISyntaxException(e);
}
}
@@ -649,7 +652,7 @@ public class HDFSOutputConnector extends
}
}
- protected void createFile(Path path, InputStream input)
+ protected void createFile(Path path, InputStream input,IOutputAddActivity activities, String documentURI)
throws ManifoldCFException, ServiceInterruption {
CreateFileThread t = new CreateFileThread(getSession(), path, input);
try {
@@ -657,13 +660,17 @@ public class HDFSOutputConnector extends
t.finishUp();
} catch (InterruptedException e) {
t.interrupt();
+ activities.recordActivity(null,INGEST_ACTIVITY,null,documentURI,activities.EXCEPTION,"Rejected due to InterruptedException.");
throw new ManifoldCFException("Interrupted: "+e.getMessage(),e,ManifoldCFException.INTERRUPTED);
} catch (java.net.SocketTimeoutException e) {
+ activities.recordActivity(null,INGEST_ACTIVITY,null,documentURI,activities.EXCEPTION,"Rejected due to SocketTimeoutException.");
handleIOException(e);
} catch (InterruptedIOException e) {
t.interrupt();
+ activities.recordActivity(null,INGEST_ACTIVITY,null,documentURI,activities.EXCEPTION,"Rejected due to InterruptedIOException.");
handleIOException(e);
} catch (IOException e) {
+ activities.recordActivity(null,INGEST_ACTIVITY,null,documentURI,activities.EXCEPTION,"Rejected due to IOException.");
handleIOException(e);
}
}
@@ -703,7 +710,7 @@ public class HDFSOutputConnector extends
}
}
- protected void deleteFile(Path path)
+ protected void deleteFile(Path path,IOutputRemoveActivity activities,String documentURI)
throws ManifoldCFException, ServiceInterruption {
// Establish a session
DeleteFileThread t = new DeleteFileThread(getSession(),path);
@@ -712,13 +719,17 @@ public class HDFSOutputConnector extends
t.finishUp();
} catch (InterruptedException e) {
t.interrupt();
+ activities.recordActivity(null,REMOVE_ACTIVITY,null,documentURI,activities.EXCEPTION,"Rejected due to InterruptedException.");
throw new ManifoldCFException("Interrupted: "+e.getMessage(),e,ManifoldCFException.INTERRUPTED);
} catch (java.net.SocketTimeoutException e) {
+ activities.recordActivity(null,REMOVE_ACTIVITY,null,documentURI,activities.EXCEPTION,"Rejected due to SocketTimeoutException.");
handleIOException(e);
} catch (InterruptedIOException e) {
t.interrupt();
+ activities.recordActivity(null,REMOVE_ACTIVITY,null,documentURI,activities.EXCEPTION,"Rejected due to InterruptedIOException.");
handleIOException(e);
} catch (IOException e) {
+ activities.recordActivity(null,REMOVE_ACTIVITY,null,documentURI,activities.EXCEPTION,"Rejected due to IOException.");
handleIOException(e);
}
}