Be more specific than IllegalArgumentException (elastic#25)

Today, we throw an overly-general IllegalArgumentException to reject
unacceptable messages in the consensus layer. We expect messages to be rejected
from time to time, if delayed or some other failure occurs, and it is not
useful to log the full stack trace leading to these exceptions.

This change introduces ConsensusMessageRejectedException for these cases, and
suppresses overly verbose logging when receiving such a message. It does not
include any of the extra machinery needed to pass these exceptions over the
wire.

Author: DaveCTurner

core/src/main/java/org/elasticsearch/ElasticsearchException.java

@@ -987,8 +987,10 @@ private enum ElasticsearchExceptionHandle {
         UNKNOWN_NAMED_OBJECT_EXCEPTION(org.elasticsearch.common.xcontent.NamedXContentRegistry.UnknownNamedObjectException.class,
                 org.elasticsearch.common.xcontent.NamedXContentRegistry.UnknownNamedObjectException::new, 148, Version.V_5_2_0),
         TOO_MANY_BUCKETS_EXCEPTION(MultiBucketConsumerService.TooManyBucketsException.class,
-                                   MultiBucketConsumerService.TooManyBucketsException::new, 149,
-            Version.V_7_0_0_alpha1);
+            MultiBucketConsumerService.TooManyBucketsException::new, 149, Version.V_7_0_0_alpha1),
+        CONSENSUS_MESSAGE_REJECTED_EXCEPTION(org.elasticsearch.discovery.zen2.ConsensusMessageRejectedException.class,
+            org.elasticsearch.discovery.zen2.ConsensusMessageRejectedException::new, 150, Version.V_7_0_0_alpha1);
+
 
         final Class<? extends ElasticsearchException> exceptionClass;
         final CheckedFunction<StreamInput, ? extends ElasticsearchException, IOException> constructor;

core/src/main/java/org/elasticsearch/discovery/zen2/ConsensusMessageRejectedException.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.discovery.zen2;
+
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.common.io.stream.StreamInput;
+
+import java.io.IOException;
+
+public class ConsensusMessageRejectedException extends ElasticsearchException {
+    public ConsensusMessageRejectedException(String msg, Object... args) {
+        super(msg, args);
+    }
+
+    public ConsensusMessageRejectedException(StreamInput in) throws IOException {
+        super(in);
+    }
+}

core/src/main/java/org/elasticsearch/discovery/zen2/ConsensusState.java

@@ -105,13 +105,13 @@ public long lastAcceptedTerm() {
      *
      * @param newTerm The new term
      * @return A Vote that must be sent to at most one other node.
-     * @throws IllegalArgumentException if the arguments were incompatible with the current state of this object.
+     * @throws ConsensusMessageRejectedException if the arguments were incompatible with the current state of this object.
      */
     public Vote handleStartVote(long newTerm) {
         if (newTerm <= getCurrentTerm()) {
             logger.debug("handleStartVote: ignored as term provided [{}] lower or equal than current term [{}]",
                 newTerm, getCurrentTerm());
-            throw new IllegalArgumentException("incoming term " + newTerm + " lower than current term " + getCurrentTerm());
+            throw new ConsensusMessageRejectedException("incoming term " + newTerm + " lower than current term " + getCurrentTerm());
         }
 
         logger.debug("handleStartVote: updating term from [{}] to [{}]", getCurrentTerm(), newTerm);
@@ -132,25 +132,26 @@ public Vote handleStartVote(long newTerm) {
      * @param sourceNode The sender of the Vote received.
      * @param vote       The Vote received.
      * @return An optional PublishRequest which, if present, can be broadcast to all peers.
-     * @throws IllegalArgumentException if the arguments were incompatible with the current state of this object.
+     * @throws ConsensusMessageRejectedException if the arguments were incompatible with the current state of this object.
      */
     public Optional<PublishRequest<T>> handleVote(DiscoveryNode sourceNode, Vote vote) {
         if (vote.getTerm() != getCurrentTerm()) {
             logger.debug("handleVote: ignored vote due to term mismatch (expected: [{}], actual: [{}])",
                 getCurrentTerm(), vote.getTerm());
-            throw new IllegalArgumentException("incoming term " + vote.getTerm() + " does not match current term " + getCurrentTerm());
+            throw new ConsensusMessageRejectedException(
+                "incoming term " + vote.getTerm() + " does not match current term " + getCurrentTerm());
         }
         if (vote.getFirstUncommittedSlot() > firstUncommittedSlot()) {
             logger.debug("handleVote: ignored vote due to slot mismatch (expected: <=[{}], actual: [{}])",
                 firstUncommittedSlot(), vote.getFirstUncommittedSlot());
-            throw new IllegalArgumentException("incoming slot " + vote.getFirstUncommittedSlot() + " higher than current slot " +
-                firstUncommittedSlot());
+            throw new ConsensusMessageRejectedException(
+                "incoming slot " + vote.getFirstUncommittedSlot() + " higher than current slot " + firstUncommittedSlot());
         }
         final long lastAcceptedTerm = lastAcceptedTerm();
         if (vote.getFirstUncommittedSlot() == firstUncommittedSlot() && vote.getLastAcceptedTerm() > lastAcceptedTerm) {
             logger.debug("handleVote: ignored vote as voter has better last accepted term (expected: <=[{}], actual: [{}])",
                 lastAcceptedTerm, vote.getLastAcceptedTerm());
-            throw new IllegalArgumentException("incoming last accepted term " + vote.getLastAcceptedTerm() + " higher than " +
+            throw new ConsensusMessageRejectedException("incoming last accepted term " + vote.getLastAcceptedTerm() + " higher than " +
                 "current last accepted term " + lastAcceptedTerm);
         }
 
@@ -177,19 +178,19 @@ public Optional<PublishRequest<T>> handleVote(DiscoveryNode sourceNode, Vote vot
      *
      * @param publishRequest The PublishRequest received.
      * @return A PublishResponse which can be sent back to the sender of the PublishRequest.
-     * @throws IllegalArgumentException if the arguments were incompatible with the current state of this object.
+     * @throws ConsensusMessageRejectedException if the arguments were incompatible with the current state of this object.
      */
     public PublishResponse handlePublishRequest(PublishRequest<T> publishRequest) {
         if (publishRequest.getTerm() != getCurrentTerm()) {
             logger.debug("handlePublishRequest: ignored publish request due to term mismatch (expected: [{}], actual: [{}])",
                 getCurrentTerm(), publishRequest.getTerm());
-            throw new IllegalArgumentException("incoming term " + publishRequest.getTerm() + " does not match current term " +
+            throw new ConsensusMessageRejectedException("incoming term " + publishRequest.getTerm() + " does not match current term " +
                 getCurrentTerm());
         }
         if (publishRequest.getSlot() != firstUncommittedSlot()) {
             logger.debug("handlePublishRequest: ignored publish request due to slot mismatch (expected: [{}], actual: [{}])",
                 firstUncommittedSlot(), publishRequest.getSlot());
-            throw new IllegalArgumentException("incoming slot " + publishRequest.getSlot() + " does not match current slot " +
+            throw new ConsensusMessageRejectedException("incoming slot " + publishRequest.getSlot() + " does not match current slot " +
                 firstUncommittedSlot());
         }
 
@@ -209,13 +210,13 @@ public PublishResponse handlePublishRequest(PublishRequest<T> publishRequest) {
      * @param publishResponse The PublishResponse received.
      * @return An optional ApplyCommit which, if present, may be broadcast to all peers, indicating that this publication
      * has been accepted at a quorum of peers and is therefore committed.
-     * @throws IllegalArgumentException if the arguments were incompatible with the current state of this object.
+     * @throws ConsensusMessageRejectedException if the arguments were incompatible with the current state of this object.
      */
     public Optional<ApplyCommit> handlePublishResponse(DiscoveryNode sourceNode, PublishResponse publishResponse) {
         if (publishResponse.getTerm() != getCurrentTerm()) {
             logger.debug("handlePublishResponse: ignored publish response due to term mismatch (expected: [{}], actual: [{}])",
                 getCurrentTerm(), publishResponse.getTerm());
-            throw new IllegalArgumentException("incoming term " + publishResponse.getTerm()
+            throw new ConsensusMessageRejectedException("incoming term " + publishResponse.getTerm()
                 + " does not match current term " + getCurrentTerm());
         }
         if (publishResponse.getSlot() != firstUncommittedSlot()) {
@@ -225,7 +226,7 @@ public Optional<ApplyCommit> handlePublishResponse(DiscoveryNode sourceNode, Pub
                 logger.debug("handlePublishResponse: ignored publish response due to slot mismatch (expected: [{}], actual: [{}])",
                     firstUncommittedSlot(), publishResponse.getSlot());
             }
-            throw new IllegalArgumentException("incoming slot " + publishResponse.getSlot() + " does not match current slot " +
+            throw new ConsensusMessageRejectedException("incoming slot " + publishResponse.getSlot() + " does not match current slot " +
                 firstUncommittedSlot());
         }
 
@@ -245,20 +246,20 @@ public Optional<ApplyCommit> handlePublishResponse(DiscoveryNode sourceNode, Pub
      * May be called on receipt of an ApplyCommit. Updates the committed state accordingly.
      *
      * @param applyCommit The ApplyCommit received.
-     * @throws IllegalArgumentException if the arguments were incompatible with the current state of this object.
+     * @throws ConsensusMessageRejectedException if the arguments were incompatible with the current state of this object.
      */
     public void handleCommit(ApplyCommit applyCommit) {
         if (applyCommit.getTerm() != lastAcceptedTerm()) {
             logger.debug("handleCommit: ignored commit request due to term mismatch " +
                     "(expected: [term {} slot {}], actual: [term {} slot {}])",
                 lastAcceptedTerm(), firstUncommittedSlot(), applyCommit.getTerm(), applyCommit.getSlot());
-            throw new IllegalArgumentException("incoming term " + applyCommit.getTerm() + " does not match last accepted term " +
+            throw new ConsensusMessageRejectedException("incoming term " + applyCommit.getTerm() + " does not match last accepted term " +
                 lastAcceptedTerm());
         }
         if (applyCommit.getSlot() != firstUncommittedSlot()) {
             logger.debug("handleCommit: ignored commit request due to slot mismatch (term {}, expected: [{}], actual: [{}])",
                 lastAcceptedTerm(), firstUncommittedSlot(), applyCommit.getSlot());
-            throw new IllegalArgumentException("incoming slot " + applyCommit.getSlot() + " does not match current slot " +
+            throw new ConsensusMessageRejectedException("incoming slot " + applyCommit.getSlot() + " does not match current slot " +
                 firstUncommittedSlot());
         }
 
@@ -295,13 +296,13 @@ public T generateCatchup() {
     /**
      * May be called on receipt of a catch-up message containing the current committed state from a peer.
      *
-     * @throws IllegalArgumentException if the arguments were incompatible with the current state of this object.
+     * @throws ConsensusMessageRejectedException if the arguments were incompatible with the current state of this object.
      */
     public void applyCatchup(T newCommittedState) {
         if (newCommittedState.getSlot() <= getCommittedState().getSlot()) {
             logger.debug("applyCatchup: ignored catch up request due to slot mismatch (expected: >[{}], actual: [{}])",
                 getCommittedState().getSlot(), newCommittedState.getSlot());
-            throw new IllegalArgumentException("incoming slot " + newCommittedState.getSlot() + " no higher than current slot " +
+            throw new ConsensusMessageRejectedException("incoming slot " + newCommittedState.getSlot() + " no higher than current slot " +
                 getCommittedState().getSlot());
         }
 
@@ -320,16 +321,16 @@ public void applyCatchup(T newCommittedState) {
      *
      * @param diff The RSM transition on which to achieve consensus.
      * @return A PublishRequest that may be broadcast to all peers.
-     * @throws IllegalArgumentException if the arguments were incompatible with the current state of this object.
+     * @throws ConsensusMessageRejectedException if the arguments were incompatible with the current state of this object.
      */
     public PublishRequest<T> handleClientValue(Diff<T> diff) {
         if (electionWon == false) {
             logger.debug("handleClientValue: ignored request as election not won");
-            throw new IllegalArgumentException("election not won");
+            throw new ConsensusMessageRejectedException("election not won");
         }
         if (publishPermitted == false) {
             logger.debug("handleClientValue: ignored request as publishing is not permitted");
-            throw new IllegalArgumentException("publishing not permitted");
+            throw new ConsensusMessageRejectedException("publishing not permitted");
         }
         assert lastAcceptedTerm() == NO_TERM; // see https://github.com/elastic/elasticsearch-formal-models/issues/24