svn commit: r992167 - in /avro/trunk: CHANGES.txt lang/java/src/java/org/apache/avro/generic/GenericDatumReader.java

[cu...@apache.org]

Author: cutting
Date: Fri Sep  3 01:27:09 2010
New Revision: 992167

URL: http://svn.apache.org/viewvc?rev=992167&view=rev
Log:
AVRO-650. Java: Fix GenericDatumReader to be thread-safe.

Modified:
    avro/trunk/CHANGES.txt
    avro/trunk/lang/java/src/java/org/apache/avro/generic/GenericDatumReader.java

Modified: avro/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/avro/trunk/CHANGES.txt?rev=992167&r1=992166&r2=992167&view=diff
==============================================================================
--- avro/trunk/CHANGES.txt (original)
+++ avro/trunk/CHANGES.txt Fri Sep  3 01:27:09 2010
@@ -233,6 +233,8 @@ Avro 1.4.0 (31 August 2010)
     AVRO-653. Python: Fix so distribution contains correct files.
     (Eric Evans via cutting)
 
+    AVRO-650. Java: Fix GenericDatumReader to be thread-safe. (cutting)
+
 Avro 1.3.3 (7 June 2010)
 
   IMPROVEMENTS

Modified: avro/trunk/lang/java/src/java/org/apache/avro/generic/GenericDatumReader.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/src/java/org/apache/avro/generic/GenericDatumReader.java?rev=992167&r1=992166&r2=992167&view=diff
==============================================================================
--- avro/trunk/lang/java/src/java/org/apache/avro/generic/GenericDatumReader.java (original)
+++ avro/trunk/lang/java/src/java/org/apache/avro/generic/GenericDatumReader.java Fri Sep  3 01:27:09 2010
@@ -36,7 +36,6 @@ import org.apache.avro.util.WeakIdentity
 public class GenericDatumReader<D> implements DatumReader<D> {
   private Schema actual;
   private Schema expected;
-  private ResolvingDecoder resolver;
 
   public GenericDatumReader() {}
 
@@ -58,14 +57,18 @@ public class GenericDatumReader<D> imple
     if (expected == null) {
       expected = actual;
     }
-    resolver = null;
+    threadResolver.set(null);
   }
 
   /** Set the reader's schema. */
   public void setExpected(Schema reader) throws IOException {
     this.expected = reader;
+    threadResolver.set(null);
   }
 
+  private final ThreadLocal<ResolvingDecoder> threadResolver =
+    new ThreadLocal<ResolvingDecoder>();
+
   private static final ThreadLocal<Map<Schema,Map<Schema,ResolvingDecoder>>>
     RESOLVER_CACHE =
     new ThreadLocal<Map<Schema,Map<Schema,ResolvingDecoder>>>() {
@@ -74,26 +77,30 @@ public class GenericDatumReader<D> imple
     }
   };
 
-  private static ResolvingDecoder getResolver(Schema actual, Schema expected)
+  private ResolvingDecoder getResolver(Schema actual, Schema expected)
     throws IOException {
+    ResolvingDecoder resolver = threadResolver.get();
+    if (resolver != null)
+      return resolver;
+
     Map<Schema,ResolvingDecoder> cache = RESOLVER_CACHE.get().get(actual);
     if (cache == null) {
       cache = new WeakIdentityHashMap<Schema,ResolvingDecoder>();
       RESOLVER_CACHE.get().put(actual, cache);
     }
-    ResolvingDecoder resolver = cache.get(expected);
+    resolver = cache.get(expected);
     if (resolver == null) {
       resolver = new ResolvingDecoder(Schema.applyAliases(actual, expected),
                                       expected, null);
       cache.put(expected, resolver);
     }
+    threadResolver.set(resolver);
     return resolver;
   }
 
   @SuppressWarnings("unchecked")
   public D read(D reuse, Decoder in) throws IOException {
-    if (resolver == null)
-      resolver = getResolver(actual, expected);
+    ResolvingDecoder resolver = getResolver(actual, expected);
     resolver.init(in);
     D result = (D) read(reuse, expected, resolver);
     resolver.drain();