Add tests and fix for concurrency problems wrt avro encoder/decoder

Author: cowtowncoder

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/AvroSchema.java

@@ -50,22 +50,26 @@ public String getSchemaType() {
 
     public static BinaryDecoder decoder(InputStream in, boolean buffering)
     {
+        /*
         SoftReference<BinaryDecoder> ref = decoderRecycler.get();
         BinaryDecoder prev = (ref == null) ? null : ref.get();
-        /* Factory will check if the decoder has a matching type for reuse.
-         * If not, it will drop the instance being reused and will return
-         * a new, proper one.
-         */
+        // Factory will check if the decoder has a matching type for reuse.
+        // If not, it will drop the instance being reused and will return a new, proper one.
         BinaryDecoder next = buffering
             ? DECODER_FACTORY.binaryDecoder(in, prev)
             : DECODER_FACTORY.directBinaryDecoder(in, prev);
 
         decoderRecycler.set(new SoftReference<BinaryDecoder>(next));
         return next;
+        */
+        return buffering
+                ? DECODER_FACTORY.binaryDecoder(in, null)
+                : DECODER_FACTORY.directBinaryDecoder(in, null);
     }
 
     public static BinaryDecoder decoder(byte[] buffer, int offset, int len)
     {
+        /*
         SoftReference<BinaryDecoder> ref = decoderRecycler.get();
         BinaryDecoder prev = (ref == null) ? null : ref.get();
         
@@ -75,25 +79,28 @@ public static BinaryDecoder decoder(byte[] buffer, int offset, int len)
         prev = DECODER_FACTORY.binaryDecoder(buffer, offset, len, null);
         decoderRecycler.set(new SoftReference<BinaryDecoder>(prev));
         return prev;
+        */
+        return DECODER_FACTORY.binaryDecoder(buffer, offset, len, null);
     }
-    
-    
+
     public static BinaryEncoder encoder(OutputStream out, boolean buffering)
     {
+        /*
         SoftReference<BinaryEncoder> ref = encoderRecycler.get();
         BinaryEncoder prev = (ref == null) ? null : ref.get();
-        /* Factory will check if the encoder has a matching type for reuse.
-         * If not, it will drop the instance being reused and will return
-         * a new, properly initialized instance
-         */
+        // Factory will check if the decoder has a matching type for reuse.
+        // If not, it will drop the instance being reused and will return a new, proper one.
         BinaryEncoder next =
             buffering
             ? ENCODER_FACTORY.binaryEncoder(out, prev)
             : ENCODER_FACTORY.directBinaryEncoder(out, prev);
         
         encoderRecycler.set(new SoftReference<BinaryEncoder>(next));
-
         return next;
+        */
+        return buffering
+            ? ENCODER_FACTORY.binaryEncoder(out, null)
+            : ENCODER_FACTORY.directBinaryEncoder(out, null);
     }
 
     public AvroStructureReader getReader()

avro/src/test/java/com/fasterxml/jackson/dataformat/avro/ConcurrencyTest.java

@@ -0,0 +1,114 @@
+package com.fasterxml.jackson.dataformat.avro;
+
+import java.io.*;
+
+import com.fasterxml.jackson.databind.MappingIterator;
+import com.fasterxml.jackson.databind.SequenceWriter;
+
+/**
+ * Simple tests for some testable aspects of concurrent usage.
+ * Specifically tries to ensure that usage of multiple parsers,
+ * generators, from same thread, does not cause problems with
+ * reuse of certain components.
+ */
+public class ConcurrencyTest extends AvroTestBase
+{
+    private final AvroMapper MAPPER = getMapper();
+
+    private final AvroSchema EMPL_SCHEMA;
+
+    public ConcurrencyTest() throws IOException {
+        EMPL_SCHEMA = getEmployeeSchema();        
+    }
+    
+    // Simple test that creates 2 encoders and uses them in interleaved manner.
+    // This should tease out simplest problems with possible encoder reuse.
+    public void testMultipleEncoders() throws Exception
+    {
+        ByteArrayOutputStream b1 = new ByteArrayOutputStream();
+        ByteArrayOutputStream b2 = new ByteArrayOutputStream();
+        SequenceWriter sw1 = MAPPER.writer(EMPL_SCHEMA)
+                .writeValues(b1);
+        SequenceWriter sw2 = MAPPER.writer(EMPL_SCHEMA)
+                .writeValues(b2);
+
+        for (int i = 0; i < 200; ++i) {
+            _writeEmpl(sw1, "foo", i);
+            _writeEmpl(sw2, "foo", i);
+        }
+        sw1.close();
+        sw2.close();
+        assertEquals(b1.size(), b2.size());
+        // value just verified once, but since Avro format stable should remain stable
+        assertEquals(6926, b1.size());
+    }
+
+    public void testMultipleDecodersBlock() throws Exception {
+        _testMultipleDecoders(false);
+    }
+
+    public void testMultipleDecodersStreaming() throws Exception {
+        _testMultipleDecoders(true);
+    }
+
+    private void _testMultipleDecoders(boolean useStream) throws Exception
+    {
+        final int ROUNDS = 40;
+        // Here let's do encoding linearly, to remove coupling with other test(s)
+        ByteArrayOutputStream b = new ByteArrayOutputStream();
+        SequenceWriter sw = MAPPER.writer(EMPL_SCHEMA)
+                .writeValues(b);
+        for (int i = 0; i < ROUNDS; ++i) {
+            _writeEmpl(sw, "a", i);
+        }
+        sw.close();
+        final byte[] b1 = b.toByteArray();
+
+        b = new ByteArrayOutputStream();
+        sw = MAPPER.writer(EMPL_SCHEMA)
+                .writeValues(b);
+        for (int i = 0; i < ROUNDS; ++i) {
+            _writeEmpl(sw, "b", i);
+        }
+        sw.close();
+        final byte[] b2 = b.toByteArray();
+
+        MappingIterator<Employee> it1, it2;
+
+        if (useStream) {
+            it1 = MAPPER.readerFor(Employee.class).with(EMPL_SCHEMA)
+                    .readValues(b1);
+            it2 = MAPPER.readerFor(Employee.class).with(EMPL_SCHEMA)
+                    .readValues(b2);
+        } else {
+            it1 = MAPPER.readerFor(Employee.class).with(EMPL_SCHEMA)
+                    .readValues(new ByteArrayInputStream(b1));
+            it2 = MAPPER.readerFor(Employee.class).with(EMPL_SCHEMA)
+                    .readValues(new ByteArrayInputStream(b2));
+        }
+
+        for (int i = 0; i < 40; ++i) {
+            assertTrue(it1.hasNextValue());
+            assertTrue(it2.hasNextValue());
+            Employee e1 = it1.nextValue();
+            Employee e2 = it2.nextValue();
+
+            assertEquals("Empl"+i+"a", e1.name);
+            assertEquals("Empl"+i+"b", e2.name);
+            assertEquals(10+i, e1.age);
+            assertEquals(10+i, e2.age);
+        }
+        assertFalse(it1.hasNextValue());
+        assertFalse(it2.hasNextValue());
+        it1.close();
+        it2.close();
+    }
+
+    private void _writeEmpl(SequenceWriter sw, String type, int index) throws IOException {
+        sw.write(_empl(type, index));
+    }
+
+    private Employee _empl(String type, int index) {
+        return new Employee("Empl"+index+type, 10+index, new String[] { "empl"+index+"@company.com" }, null);
+    }
+}

avro/src/test/java/com/fasterxml/jackson/dataformat/avro/RootSequenceTest.java

@@ -79,7 +79,7 @@ public void testReadWriteEmployeeSequence() throws Exception
         Employee peon2 = new Employee("Worker#2", 42, new String[] { "worker2@company.com" }, boss);
 
         // First: write a sequence of 3 root-level Employee Objects
-        
+
         SequenceWriter sw = MAPPER.writerFor(Employee.class)
                 .with(getEmployeeSchema())
                 .writeValues(b);