Assert names of read writeables

Adds an assertion that when reading a NamedWriteable it has the same name
you read. It'd be *super* weird if it didn't.

core/src/main/java/org/elasticsearch/common/io/stream/NamedWriteableAwareStreamInput.java

@@ -34,7 +34,7 @@ public class NamedWriteableAwareStreamInput extends FilterStreamInput {
     }
 
     @Override
-    <C> C readNamedWriteable(Class<C> categoryClass) throws IOException {
+    <C extends NamedWriteable<?>> C readNamedWriteable(Class<C> categoryClass) throws IOException {
         String name = readString();
         Writeable.Reader<? extends C> reader = namedWriteableRegistry.getReader(categoryClass, name);
         C c = reader.read(this);
@@ -42,6 +42,8 @@ public class NamedWriteableAwareStreamInput extends FilterStreamInput {
             throw new IOException(
                     "Writeable.Reader [" + reader + "] returned null which is not allowed and probably means it screwed up the stream.");
         }
+        assert name.equals(c.getWriteableName()) : c + " claims to have a different name [" + c.getWriteableName()
+                + "] than it was read from [" + name + "].";
         return c;
     }
 }

core/src/main/java/org/elasticsearch/common/io/stream/StreamInput.java

@@ -707,7 +707,7 @@ public abstract class StreamInput extends InputStream {
      * Default implementation throws {@link UnsupportedOperationException} as StreamInput doesn't hold a registry.
      * Use {@link FilterInputStream} instead which wraps a stream and supports a {@link NamedWriteableRegistry} too.
      */
-    <C> C readNamedWriteable(@SuppressWarnings("unused") Class<C> categoryClass) throws IOException {
+    <C extends NamedWriteable<?>> C readNamedWriteable(@SuppressWarnings("unused") Class<C> categoryClass) throws IOException {
         throw new UnsupportedOperationException("can't read named writeable from StreamInput");
     }
 

core/src/test/java/org/elasticsearch/common/io/stream/BytesStreamsTests.java

@@ -411,6 +411,25 @@ public class BytesStreamsTests extends ESTestCase {
         assertThat(e.getMessage(), endsWith("] returned null which is not allowed and probably means it screwed up the stream."));
     }
 
+    public void testWriteableReaderReturnsWrongName() throws IOException {
+        BytesStreamOutput out = new BytesStreamOutput();
+        NamedWriteableRegistry namedWriteableRegistry = new NamedWriteableRegistry();
+        namedWriteableRegistry.register(BaseNamedWriteable.class, TestNamedWriteable.NAME, (StreamInput in) -> new TestNamedWriteable(in) {
+            @Override
+            public String getWriteableName() {
+                return "intentionally-broken";
+            }
+        });
+        TestNamedWriteable namedWriteableIn = new TestNamedWriteable(randomAsciiOfLengthBetween(1, 10), randomAsciiOfLengthBetween(1, 10));
+        out.writeNamedWriteable(namedWriteableIn);
+        byte[] bytes = out.bytes().toBytes();
+        StreamInput in = new NamedWriteableAwareStreamInput(StreamInput.wrap(bytes), namedWriteableRegistry);
+        assertEquals(in.available(), bytes.length);
+        AssertionError e = expectThrows(AssertionError.class, () -> in.readNamedWriteable(BaseNamedWriteable.class));
+        assertThat(e.getMessage(),
+                endsWith(" claims to have a different name [intentionally-broken] than it was read from [test-named-writeable]."));
+    }
+
     private static abstract class BaseNamedWriteable<T> implements NamedWriteable<T> {
 
     }