AVRO-2302 Fix invalid namespace on BigQuery import

This change fixes an issue when importing Avro files into BigQuery, when
such files have been generated using `ProtobufData`. The schema
namespace may end in a dollar sign, if using nested classes.

The correction piggybacks on the fix provided for AVRO-2143 (apache#283). In
it, `SpecificData` will attempt to load a class joining the namespace
and the name with a dot, and if that fails, it tries again with a dollar
sign.

The same behaviour is hereby reused for `ProtobufData`.

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

@@ -45,7 +45,6 @@
 import com.google.protobuf.Descriptors.FileDescriptor;
 import com.google.protobuf.DescriptorProtos.FileOptions;
 
-import org.apache.avro.util.ClassUtils;
 import org.apache.avro.util.internal.Accessor;
 import com.fasterxml.jackson.core.JsonFactory;
 import com.fasterxml.jackson.databind.JsonNode;
@@ -135,7 +134,7 @@ protected boolean isRecord(Object datum) {
   @Override
   public Object newRecord(Object old, Schema schema) {
     try {
-      Class c = ClassUtils.forName(SpecificData.getClassName(schema));
+      Class c = SpecificData.get().getClass(schema);
       if (c == null)
         return newRecord(old, schema);            // punt to generic
       if (c.isInstance(old))
@@ -239,10 +238,10 @@ private String getNamespace(FileDescriptor fd, Descriptor containing) {
     }
     String inner = "";
     while (containing != null) {
-      inner = containing.getName() + "$" + inner;
+      inner = "$" + containing.getName() + inner;
       containing = containing.getContainingType();
     }
-    return p + "." + outer + "$" + inner;
+    return p + "." + outer + inner;
   }
 
   private static String toCamelCase(String s){

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

@@ -25,7 +25,6 @@
 import org.apache.avro.specific.SpecificData;
 import org.apache.avro.io.Decoder;
 import org.apache.avro.io.ResolvingDecoder;
-import org.apache.avro.util.ClassUtils;
 
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Message;
@@ -67,7 +66,7 @@ protected Object readRecord(Object old, Schema expected,
   @Override
   protected Object createEnum(String symbol, Schema schema) {
     try {
-      Class c = ClassUtils.forName(SpecificData.getClassName(schema));
+      Class c = SpecificData.get().getClass(schema);
       if (c == null) return super.createEnum(symbol, schema); // punt to generic
       return ((ProtocolMessageEnum)Enum.valueOf(c, symbol)).getValueDescriptor();
     } catch (Exception e) {

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

@@ -86,7 +86,11 @@ public class TestProtobuf {
 
   @Test public void testNestedEnum() throws Exception {
     Schema s = ProtobufData.get().getSchema(N.class);
-    assertEquals(N.class.getName(), SpecificData.getClassName(s));
+    assertEquals(N.class.getName(), SpecificData.get().getClass(s).getName());
   }
 
+  @Test public void testNestedClassNamespace() throws Exception {
+    Schema s = ProtobufData.get().getSchema(Foo.class);
+    assertEquals(org.apache.avro.protobuf.Test.class.getName(), s.getNamespace());
+  }
 }