Replace '$' in namespace with __dollar__.

lang/java/avro/src/main/java/org/apache/avro/Schema.java

@@ -717,10 +717,14 @@ public Name(String name, String space) {
         space = name.substring(0, lastDot); // get space from name
         this.name = validateName(name.substring(lastDot + 1));
       }
-      this.space = validateSpace(space);
+      this.space = "".equals(space) || space == null ? null : validateSpace(space);
       this.full = (this.space == null) ? this.name : this.space + "." + this.name;
     }
 
+    protected String validateSpace(String space) {
+      return Schema.validateSpace(space);
+    }
+
     @Override
     public boolean equals(Object o) {
       if (o == this)
@@ -781,6 +785,18 @@ private boolean shouldWriteFull(String defaultSpace) {
 
   }
 
+  static class Alias extends Name {
+
+    public Alias(String name, String space) {
+      super(name, space);
+    }
+
+    @Override
+    protected String validateSpace(String space) {
+      return space;
+    }
+  }
+
   private static abstract class NamedSchema extends Schema {
     final Name name;
     final String doc;
@@ -826,7 +842,7 @@ public void addAlias(String name, String space) {
         this.aliases = new LinkedHashSet<>();
       if (space == null)
         space = this.name.space;
-      aliases.add(new Name(name, space));
+      aliases.add(new Alias(name, space));
     }
 
     @Override
@@ -1644,10 +1660,6 @@ private static String validateName(String name) {
   }
 
   private static String validateSpace(String space) {
-    if ("".equals(space) || space == null) {
-      return null;
-    }
-
     for (String part : space.split("\\.")) {
       validateName(part);
     }

lang/java/avro/src/main/java/org/apache/avro/specific/SpecificData.java

@@ -53,6 +53,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.function.Function;
+import java.util.regex.Matcher;
 
 /** Utilities for generated Java classes and interfaces. */
 public class SpecificData extends GenericData {
@@ -255,6 +256,7 @@ public Class getClass(Schema schema) {
       String name = schema.getFullName();
       if (name == null)
         return null;
+      name = name.replaceAll("__dollar__", Matcher.quoteReplacement("$"));
       Class<?> c = MapUtil.computeIfAbsent(classCache, name, n -> {
         try {
           return ClassUtils.forName(getClassLoader(), getClassName(schema));
@@ -332,6 +334,7 @@ public static String getClassName(Schema schema) {
     if (namespace == null || "".equals(namespace))
       return name;
 
+    namespace = namespace.replaceAll("__dollar__", Matcher.quoteReplacement("$"));
     StringBuilder classNameBuilder = new StringBuilder();
     String[] words = namespace.split("\\.");
 

lang/java/protobuf/src/main/java/org/apache/avro/protobuf/ProtobufData.java

@@ -242,12 +242,12 @@ public String getNamespace(FileDescriptor fd, Descriptor containing) {
       if (inner.length() == 0) {
         inner.insert(0, containing.getName());
       } else {
-        inner.insert(0, containing.getName() + "$");
+        inner.insert(0, containing.getName() + "__dollar__");
       }
       containing = containing.getContainingType();
     }
     String d1 = (!outer.isEmpty() || inner.length() != 0 ? "." : "");
-    String d2 = (!outer.isEmpty() && inner.length() != 0 ? "$" : "");
+    String d2 = (!outer.isEmpty() && inner.length() != 0 ? "__dollar__" : "");
     return p + d1 + outer + d2 + inner;
   }
 

lang/java/protobuf/src/test/java/org/apache/avro/protobuf/TestProtobuf.java

@@ -32,9 +32,11 @@
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 
 import com.google.protobuf.ByteString;
+import com.google.protobuf.Descriptors.EnumValueDescriptor;
 
 import org.apache.avro.protobuf.noopt.Test.Foo;
 import org.apache.avro.protobuf.noopt.Test.A;
+import org.apache.avro.protobuf.noopt.Test.Bar.BarInner;
 import org.apache.avro.protobuf.noopt.Test.M.N;
 
 public class TestProtobuf {
@@ -122,6 +124,30 @@ void nestedClassNamespace() throws Exception {
     assertEquals(org.apache.avro.protobuf.noopt.Test.class.getName(), s.getNamespace());
   }
 
+  @Test
+  void nestedClassWithDollarSymbol() throws Exception {
+    BarInner.Builder builder = BarInner.newBuilder();
+    builder.setA(1);
+    BarInner record = builder.build();
+
+    ByteArrayOutputStream bao = new ByteArrayOutputStream();
+    ProtobufDatumWriter<BarInner> w = new ProtobufDatumWriter<>(BarInner.class);
+    Encoder e = EncoderFactory.get().binaryEncoder(bao, null);
+    w.write(record, e);
+    e.flush();
+
+    Object o = new ProtobufDatumReader<>(BarInner.class).read(null,
+        DecoderFactory.get().binaryDecoder(new ByteArrayInputStream(bao.toByteArray()), null));
+    assertEquals(record, o);
+
+    Schema schema = ProtobufData.get().getSchema(BarInner.class);
+    assertEquals("org.apache.avro.protobuf.noopt.Test__dollar__Bar", schema.getNamespace());
+
+    // Ensure aliases accept "$" and no exception is thrown for backward
+    // compatibility.
+    schema.addAlias("org.apache.avro.protobuf.noopt.Test$Bar.BarInner");
+  }
+
   @Test
   void classNamespaceInMultipleFiles() throws Exception {
     Schema fooSchema = ProtobufData.get().getSchema(org.apache.avro.protobuf.multiplefiles.Foo.class);