[Java] Improve ProtoCoder decode error for raw Message type

CHANGES.md

@@ -43,6 +43,7 @@
 
 ## Bugfixes
 
+* (Java) Improved the error message when decoding with a ProtoCoder for the raw `com.google.protobuf.Message` type (for example inferred from a generic `T extends Message`): it now explains how to fix it instead of surfacing a `NoSuchMethodException` ([#19713](https://github.com/apache/beam/issues/19713)).
 * Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 
 ## Security Fixes

sdks/java/extensions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoCoder.java

@@ -309,7 +309,16 @@ protected synchronized Parser<T> getParser() {
           Parser<T> tParser = (Parser<T>) protoMessageInstance.getParserForType();
           memoizedParser = tParser;
         }
-      } catch (IllegalAccessException | InvocationTargetException | NoSuchMethodException e) {
+      } catch (NoSuchMethodException e) {
+        throw new IllegalArgumentException(
+            String.format(
+                "Cannot decode %s: it has no static getDefaultInstance() method, so its Parser "
+                    + "cannot be determined. This happens for the raw %s interface, e.g. when the "
+                    + "coder was inferred for a generic 'T extends Message'. Use a concrete "
+                    + "generated message class, DynamicProtoCoder, or call setCoder().",
+                protoMessageClass.getName(), Message.class.getName()),
+            e);
+      } catch (IllegalAccessException | InvocationTargetException e) {
         throw new IllegalArgumentException(e);
       }
     }

sdks/java/extensions/protobuf/src/test/java/org/apache/beam/sdk/extensions/protobuf/ProtoCoderTest.java

@@ -17,9 +17,14 @@
  */
 package org.apache.beam.sdk.extensions.protobuf;
 
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
 
+import com.google.protobuf.Message;
+import java.io.ByteArrayInputStream;
 import java.io.ObjectStreamClass;
 import java.util.Collections;
 import org.apache.beam.sdk.coders.CannotProvideCoderException;
@@ -180,4 +185,28 @@ public void testSerialVersionID() {
     long serialVersionID = ObjectStreamClass.lookup(ProtoCoder.class).getSerialVersionUID();
     assertEquals(-5043999806040629525L, serialVersionID);
   }
+
+  /** Decoding with a coder for the raw {@code Message} interface must fail with guidance. */
+  @Test
+  @SuppressWarnings({"unchecked", "rawtypes"})
+  public void testDecodeWithRawMessageClass() {
+    ProtoCoder coder = ProtoCoder.of((Class) Message.class);
+    byte[] bytes = MessageA.newBuilder().setField1("hello").build().toByteArray();
+
+    IllegalArgumentException e =
+        assertThrows(
+            IllegalArgumentException.class, () -> coder.decode(new ByteArrayInputStream(bytes)));
+    assertTrue(e.getMessage(), e.getMessage().contains("DynamicProtoCoder"));
+  }
+
+  /** Encoding never consults the parser, so a raw-{@code Message} coder can still encode. */
+  @Test
+  @SuppressWarnings({"unchecked", "rawtypes"})
+  public void testEncodeWithRawMessageClass() throws Exception {
+    MessageA value = MessageA.newBuilder().setField1("hello").build();
+    ProtoCoder coder = ProtoCoder.of((Class) Message.class);
+    assertArrayEquals(
+        CoderUtils.encodeToByteArray(ProtoCoder.of(MessageA.class), value),
+        CoderUtils.encodeToByteArray(coder, value));
+  }
 }