Remove ShardRejectedOperationException which is unused. Catch ShardOfflin

src/main/scala/com/twitter/gizzard/scheduler/JobScheduler.scala

@@ -8,7 +8,7 @@ import com.twitter.util.Duration
 import com.twitter.conversions.time._
 import com.twitter.logging.Logger
 import com.twitter.gizzard.Stats
-import com.twitter.gizzard.shards.{ShardBlackHoleException, ShardRejectedOperationException}
+import com.twitter.gizzard.shards.{ShardBlackHoleException, ShardOfflineException}
 import com.twitter.gizzard.util.Process
 
 
@@ -152,8 +152,8 @@ extends Process with JobConsumer {
           Stats.incr("job-success-count")
         } catch {
           case e: ShardBlackHoleException => Stats.incr("job-blackholed-count")
-          case e: ShardRejectedOperationException =>
-            Stats.incr("job-darkmoded-count")
+          case e: ShardOfflineException =>
+            Stats.incr("job-blocked-count")
             errorQueue.put(job)
           case e =>
             Stats.incr("job-error-count")

src/main/scala/com/twitter/gizzard/shards/ShardException.scala

@@ -41,15 +41,6 @@ class ShardDatabaseTimeoutException(timeout: Duration, shardId: ShardId, cause:
   def this(timeout: Duration, shardId: ShardId) = this(timeout, shardId, null)
 }
 
-/**
- * Shard refused to do the operation, possibly because it's blocked. This is not a retryable error.
- *
- * Often this exception is used to signal a ReplicatingShard that it should try another shard,
- * because this shard is read-only, write-only, or blocked (offline).
- */
-class ShardRejectedOperationException(description: String, shardId: ShardId) extends
-  NormalShardException(description, shardId)
-
 /**
  * Shard cannot do the operation because all possible child shards are unavailable. This is only
  * thrown by a ReplicatingShard. This is not a retryable error.

src/main/scala/net/lag/kestrel/Journal.scala

@@ -209,6 +209,7 @@ class Journal(queuePath: String, queueName: String, syncJournal: => Boolean, mul
             f(item)
           case (JournalItem.EndOfFile, _) =>
             // move to next file and try again.
+            rj.close
             readerFilename = Journal.journalAfter(new File(queuePath), queueName, readerFilename.get)
             reader = Some(new FileInputStream(new File(queuePath, readerFilename.get)).getChannel)
             fillReadBehind(f)

src/test/scala/com/twitter/gizzard/scheduler_new/JobSchedulerSpec.scala

@@ -148,11 +148,11 @@ class JobSchedulerSpec extends ConfiguredSpecification with JMocker with ClassMo
         jobScheduler.process()
       }
 
-      "darkmode" in {
+      "blocked" in {
         expect {
           one(queue).get() willReturn Some(ticket1)
           one(ticket1).job willReturn job1
-          one(job1).apply() willThrow new ShardRejectedOperationException("darkmoded!", shardId)
+          one(job1).apply() willThrow new ShardOfflineException(shardId)
           one(ticket1).ack()
           one(job1).nextJob willReturn None
           one(errorQueue).put(job1)