You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by pt...@apache.org on 2016/07/14 20:30:46 UTC
[2/4] storm git commit: STORM-1910 - Fixing if () stmts as per coding
convention
STORM-1910 - Fixing if () stmts as per coding convention
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/825547a2
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/825547a2
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/825547a2
Branch: refs/heads/1.x-branch
Commit: 825547a298e92c4f5fdbc1a3e6e005753eb5f05e
Parents: d3ec602
Author: Roshan Naik <ro...@hortonworks.com>
Authored: Thu Jul 14 11:56:41 2016 -0700
Committer: Roshan Naik <ro...@hortonworks.com>
Committed: Thu Jul 14 11:56:41 2016 -0700
----------------------------------------------------------------------
.../org/apache/storm/hdfs/spout/HdfsSpout.java | 88 ++++++++++----------
1 file changed, 44 insertions(+), 44 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/825547a2/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/HdfsSpout.java
----------------------------------------------------------------------
diff --git a/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/HdfsSpout.java b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/HdfsSpout.java
index b6815dd..6623a3e 100644
--- a/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/HdfsSpout.java
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/HdfsSpout.java
@@ -198,7 +198,7 @@ public class HdfsSpout extends BaseRichSpout {
return;
}
- if( ackEnabled && tracker.size()>= maxOutstanding) {
+ if ( ackEnabled && tracker.size()>= maxOutstanding ) {
LOG.warn("Waiting for more ACKs before generating new tuples. " +
"Progress tracker size has reached limit {}, SpoutID {}"
, maxOutstanding, spoutId);
@@ -219,7 +219,7 @@ public class HdfsSpout extends BaseRichSpout {
fileReadCompletely=false;
}
}
- if( fileReadCompletely ) { // wait for more ACKs before proceeding
+ if ( fileReadCompletely ) { // wait for more ACKs before proceeding
return;
}
// 4) Read record from file, emit to collector and record progress
@@ -230,7 +230,7 @@ public class HdfsSpout extends BaseRichSpout {
MessageId msgId = new MessageId(tupleCounter, reader.getFilePath(), reader.getFileOffset());
emitData(tuple, msgId);
- if(!ackEnabled) {
+ if ( !ackEnabled ) {
++acksSinceLastCommit; // assume message is immediately ACKed in non-ack mode
commitProgress(reader.getFileOffset());
} else {
@@ -239,7 +239,7 @@ public class HdfsSpout extends BaseRichSpout {
return;
} else {
fileReadCompletely = true;
- if(!ackEnabled) {
+ if ( !ackEnabled ) {
markFileAsDone(reader.getFilePath());
}
}
@@ -260,7 +260,7 @@ public class HdfsSpout extends BaseRichSpout {
// will commit progress into lock file if commit threshold is reached
private void commitProgress(FileOffset position) {
- if(position==null) {
+ if ( position==null ) {
return;
}
if ( lock!=null && canCommitNow() ) {
@@ -278,7 +278,7 @@ public class HdfsSpout extends BaseRichSpout {
}
private void setupCommitElapseTimer() {
- if(commitFrequencySec<=0) {
+ if ( commitFrequencySec<=0 ) {
return;
}
TimerTask timerTask = new TimerTask() {
@@ -334,7 +334,7 @@ public class HdfsSpout extends BaseRichSpout {
private static void releaseLockAndLog(FileLock fLock, String spoutId) {
try {
- if(fLock!=null) {
+ if ( fLock!=null ) {
fLock.release();
LOG.debug("Spout {} released FileLock. SpoutId = {}", fLock.getLockFile(), spoutId);
}
@@ -361,10 +361,10 @@ public class HdfsSpout extends BaseRichSpout {
this.tupleCounter = 0;
// Hdfs related settings
- if(this.hdfsUri==null && conf.containsKey(Configs.HDFS_URI)) {
+ if ( this.hdfsUri==null && conf.containsKey(Configs.HDFS_URI) ) {
this.hdfsUri = conf.get(Configs.HDFS_URI).toString();
}
- if(this.hdfsUri==null) {
+ if ( this.hdfsUri==null ) {
throw new RuntimeException("HDFS Uri not set on spout");
}
@@ -378,7 +378,7 @@ public class HdfsSpout extends BaseRichSpout {
if ( conf.containsKey(configKey) ) {
Map<String, Object> map = (Map<String, Object>)conf.get(configKey);
- if(map != null) {
+ if ( map != null ) {
for(String keyName : map.keySet()){
LOG.info("HDFS Config override : {} = {} ", keyName, String.valueOf(map.get(keyName)));
this.hdfsConfig.set(keyName, String.valueOf(map.get(keyName)));
@@ -389,11 +389,11 @@ public class HdfsSpout extends BaseRichSpout {
LOG.error("HDFS Login failed ", e);
throw new RuntimeException(e);
}
- } // if(map != null)
+ } // if (map != null)
}
// Reader type config
- if( readerType==null && conf.containsKey(Configs.READER_TYPE) ) {
+ if ( readerType==null && conf.containsKey(Configs.READER_TYPE) ) {
readerType = conf.get(Configs.READER_TYPE).toString();
}
checkValidReader(readerType);
@@ -402,7 +402,7 @@ public class HdfsSpout extends BaseRichSpout {
if ( sourceDir==null && conf.containsKey(Configs.SOURCE_DIR) ) {
sourceDir = conf.get(Configs.SOURCE_DIR).toString();
}
- if( sourceDir==null ) {
+ if ( sourceDir==null ) {
LOG.error(Configs.SOURCE_DIR + " setting is required");
throw new RuntimeException(Configs.SOURCE_DIR + " setting is required");
}
@@ -423,7 +423,7 @@ public class HdfsSpout extends BaseRichSpout {
if ( badFilesDir==null && conf.containsKey(Configs.BAD_DIR) ) {
badFilesDir = conf.get(Configs.BAD_DIR).toString();
}
- if(badFilesDir==null) {
+ if ( badFilesDir==null ) {
LOG.error(Configs.BAD_DIR + " setting is required");
throw new RuntimeException(Configs.BAD_DIR + " setting is required");
}
@@ -436,10 +436,10 @@ public class HdfsSpout extends BaseRichSpout {
}
// -- lock dir config
- if( lockDir==null && conf.containsKey(Configs.LOCK_DIR) ) {
+ if ( lockDir==null && conf.containsKey(Configs.LOCK_DIR) ) {
lockDir = conf.get(Configs.LOCK_DIR).toString();
}
- if(lockDir==null) {
+ if ( lockDir==null ) {
lockDir = getDefaultLockDir(sourceDirPath);
}
this.lockDirPath = new Path(lockDir);
@@ -447,13 +447,13 @@ public class HdfsSpout extends BaseRichSpout {
// -- lock timeout
- if( conf.get(Configs.LOCK_TIMEOUT) !=null ) {
+ if ( conf.get(Configs.LOCK_TIMEOUT) !=null ) {
this.lockTimeoutSec = Integer.parseInt(conf.get(Configs.LOCK_TIMEOUT).toString());
}
// -- enable/disable ACKing
Object ackers = conf.get(Config.TOPOLOGY_ACKER_EXECUTORS);
- if( ackers!=null ) {
+ if ( ackers!=null ) {
int ackerCount = Integer.parseInt(ackers.toString());
this.ackEnabled = (ackerCount>0);
LOG.debug("ACKer count = {}", ackerCount);
@@ -466,25 +466,25 @@ public class HdfsSpout extends BaseRichSpout {
LOG.info("ACK mode is {}", ackEnabled ? "enabled" : "disabled");
// -- commit frequency - count
- if( conf.get(Configs.COMMIT_FREQ_COUNT) != null ) {
+ if ( conf.get(Configs.COMMIT_FREQ_COUNT) != null ) {
commitFrequencyCount = Integer.parseInt(conf.get(Configs.COMMIT_FREQ_COUNT).toString());
}
// -- commit frequency - seconds
- if( conf.get(Configs.COMMIT_FREQ_SEC) != null ) {
+ if ( conf.get(Configs.COMMIT_FREQ_SEC) != null ) {
commitFrequencySec = Integer.parseInt(conf.get(Configs.COMMIT_FREQ_SEC).toString());
- if(commitFrequencySec<=0) {
+ if ( commitFrequencySec<=0 ) {
throw new RuntimeException(Configs.COMMIT_FREQ_SEC + " setting must be greater than 0");
}
}
// -- max outstanding tuples
- if( conf.get(Configs.MAX_OUTSTANDING) !=null ) {
+ if ( conf.get(Configs.MAX_OUTSTANDING) !=null ) {
maxOutstanding = Integer.parseInt(conf.get(Configs.MAX_OUTSTANDING).toString());
}
// -- clocks in sync
- if( conf.get(Configs.CLOCKS_INSYNC) !=null ) {
+ if ( conf.get(Configs.CLOCKS_INSYNC) !=null ) {
clocksInSync = Boolean.parseBoolean(conf.get(Configs.CLOCKS_INSYNC).toString());
}
@@ -497,12 +497,12 @@ public class HdfsSpout extends BaseRichSpout {
private static void validateOrMakeDir(FileSystem fs, Path dir, String dirDescription) {
try {
- if(fs.exists(dir)) {
- if(! fs.isDirectory(dir) ) {
+ if ( fs.exists(dir) ) {
+ if ( !fs.isDirectory(dir) ) {
LOG.error(dirDescription + " directory is a file, not a dir. " + dir);
throw new RuntimeException(dirDescription + " directory is a file, not a dir. " + dir);
}
- } else if(! fs.mkdirs(dir) ) {
+ } else if ( ! fs.mkdirs(dir) ) {
LOG.error("Unable to create " + dirDescription + " directory " + dir);
throw new RuntimeException("Unable to create " + dirDescription + " directory " + dir);
}
@@ -517,7 +517,7 @@ public class HdfsSpout extends BaseRichSpout {
}
private static void checkValidReader(String readerType) {
- if(readerType.equalsIgnoreCase(Configs.TEXT) || readerType.equalsIgnoreCase(Configs.SEQ) )
+ if ( readerType.equalsIgnoreCase(Configs.TEXT) || readerType.equalsIgnoreCase(Configs.SEQ) )
return;
try {
Class<?> classType = Class.forName(readerType);
@@ -535,7 +535,7 @@ public class HdfsSpout extends BaseRichSpout {
@Override
public void ack(Object msgId) {
LOG.trace("Ack received for msg {} on spout {}", msgId, spoutId);
- if(!ackEnabled) {
+ if ( !ackEnabled ) {
return;
}
MessageId id = (MessageId) msgId;
@@ -543,7 +543,7 @@ public class HdfsSpout extends BaseRichSpout {
++acksSinceLastCommit;
tracker.recordAckedOffset(id.offset);
commitProgress(tracker.getCommitPosition());
- if(fileReadCompletely && inflight.isEmpty()) {
+ if ( fileReadCompletely && inflight.isEmpty() ) {
markFileAsDone(reader.getFilePath());
reader = null;
}
@@ -552,7 +552,7 @@ public class HdfsSpout extends BaseRichSpout {
private boolean canCommitNow() {
- if( commitFrequencyCount>0 && acksSinceLastCommit >= commitFrequencyCount ) {
+ if ( commitFrequencyCount>0 && acksSinceLastCommit >= commitFrequencyCount ) {
return true;
}
return commitTimeElapsed.get();
@@ -562,7 +562,7 @@ public class HdfsSpout extends BaseRichSpout {
public void fail(Object msgId) {
LOG.trace("Fail received for msg id {} on spout {}", msgId, spoutId);
super.fail(msgId);
- if(ackEnabled) {
+ if ( ackEnabled ) {
HdfsUtils.Pair<MessageId, List<Object>> item = HdfsUtils.Pair.of(msgId, inflight.remove(msgId));
retryList.add(item);
}
@@ -572,7 +572,7 @@ public class HdfsSpout extends BaseRichSpout {
try {
// 1) If there are any abandoned files, pick oldest one
lock = getOldestExpiredLock();
- if (lock != null) {
+ if ( lock!=null ) {
LOG.debug("Spout {} now took over ownership of abandoned FileLock {}", spoutId, lock.getLockFile());
Path file = getFileForLockFile(lock.getLockFile(), sourceDirPath);
String resumeFromOffset = lock.getLastLogEntry().fileOffset;
@@ -641,20 +641,20 @@ public class HdfsSpout extends BaseRichSpout {
}
// 3 - if clocks are not in sync ..
- if( lastExpiredLock == null ) {
+ if ( lastExpiredLock == null ) {
// just make a note of the oldest expired lock now and check if its still unmodified after lockTimeoutSec
lastExpiredLock = FileLock.locateOldestExpiredLock(hdfs, lockDirPath, lockTimeoutSec);
lastExpiredLockTime = System.currentTimeMillis();
return null;
}
// see if lockTimeoutSec time has elapsed since we last selected the lock file
- if( hasExpired(lastExpiredLockTime) ) {
+ if ( hasExpired(lastExpiredLockTime) ) {
return null;
}
// If lock file has expired, then own it
FileLock.LogEntry lastEntry = FileLock.getLastEntry(hdfs, lastExpiredLock.getKey());
- if( lastEntry.equals(lastExpiredLock.getValue()) ) {
+ if ( lastEntry.equals(lastExpiredLock.getValue()) ) {
FileLock result = FileLock.takeOwnership(hdfs, lastExpiredLock.getKey(), lastEntry, spoutId);
lastExpiredLock = null;
return result;
@@ -681,10 +681,10 @@ public class HdfsSpout extends BaseRichSpout {
*/
private FileReader createFileReader(Path file)
throws IOException {
- if(readerType.equalsIgnoreCase(Configs.SEQ)) {
+ if ( readerType.equalsIgnoreCase(Configs.SEQ) ) {
return new SequenceFileReader(this.hdfs, file, conf);
}
- if(readerType.equalsIgnoreCase(Configs.TEXT)) {
+ if ( readerType.equalsIgnoreCase(Configs.TEXT) ) {
return new TextFileReader(this.hdfs, file, conf);
}
try {
@@ -707,10 +707,10 @@ public class HdfsSpout extends BaseRichSpout {
*/
private FileReader createFileReader(Path file, String offset)
throws IOException {
- if(readerType.equalsIgnoreCase(Configs.SEQ)) {
+ if ( readerType.equalsIgnoreCase(Configs.SEQ) ) {
return new SequenceFileReader(this.hdfs, file, conf, offset);
}
- if(readerType.equalsIgnoreCase(Configs.TEXT)) {
+ if ( readerType.equalsIgnoreCase(Configs.TEXT) ) {
return new TextFileReader(this.hdfs, file, conf, offset);
}
@@ -749,11 +749,11 @@ public class HdfsSpout extends BaseRichSpout {
throws IOException {
String lockFileName = lockFile.getName();
Path dataFile = new Path(sourceDirPath + Path.SEPARATOR + lockFileName + inprogress_suffix);
- if( hdfs.exists(dataFile) ) {
+ if ( hdfs.exists(dataFile) ) {
return dataFile;
}
dataFile = new Path(sourceDirPath + Path.SEPARATOR + lockFileName);
- if(hdfs.exists(dataFile)) {
+ if ( hdfs.exists(dataFile) ) {
return dataFile;
}
return null;
@@ -768,7 +768,7 @@ public class HdfsSpout extends BaseRichSpout {
Path newFile = new Path( archiveDirPath + Path.SEPARATOR + newName );
LOG.info("Completed consuming file {}", fileNameMinusSuffix);
- if (!hdfs.rename(file, newFile) ) {
+ if ( !hdfs.rename(file, newFile) ) {
throw new IOException("Rename failed for file: " + file);
}
LOG.debug("Renamed file {} to {} ", file, newFile);
@@ -797,10 +797,10 @@ public class HdfsSpout extends BaseRichSpout {
@Override
public int compareTo(MessageId rhs) {
- if (msgNumber<rhs.msgNumber) {
+ if ( msgNumber<rhs.msgNumber ) {
return -1;
}
- if(msgNumber>rhs.msgNumber) {
+ if ( msgNumber>rhs.msgNumber ) {
return 1;
}
return 0;