Fixed ContentData entity cleanup when null value repeatedly applied to cm:content property

git-svn-id: https://svn.alfresco.com/repos/alfresco-enterprise/alfresco/HEAD/root@14826 c4b6b30b-aa2e-2d43-bbcb-ca4b014f7261

source/java/org/alfresco/repo/content/AbstractContentAccessor.java

@@ -221,7 +221,9 @@ public abstract class AbstractContentAccessor implements ContentAccessor
 
     /**
      * Advise that listens for the completion of specific methods on the
-     * {@link java.nio.channels.ByteChannel} interface.
+     * {@link java.nio.channels.ByteChannel} interface.  This advise reacts
+     * only in the {@link #afterReturning(Object, Method, Object[], Object) afterReturning} phase
+     * so that the underlying stream methods have been successfully completed.
      * 
      * @author Derek Hulley
      */

source/java/org/alfresco/repo/content/RoutingContentServiceTest.java

@@ -40,6 +40,7 @@ import org.alfresco.repo.content.transform.ContentTransformer;
 import org.alfresco.repo.policy.JavaBehaviour;
 import org.alfresco.repo.policy.PolicyComponent;
 import org.alfresco.repo.security.authentication.AuthenticationComponent;
+import org.alfresco.repo.transaction.RetryingTransactionHelper.RetryingTransactionCallback;
 import org.alfresco.service.ServiceRegistry;
 import org.alfresco.service.cmr.repository.ChildAssociationRef;
 import org.alfresco.service.cmr.repository.ContentData;
@@ -73,6 +74,7 @@ public class RoutingContentServiceTest extends TestCase
         
     private static final String TEST_NAMESPACE = "http://www.alfresco.org/test/RoutingContentServiceTest";
     
+    private TransactionService transactionService;
     private ContentService contentService;
     private PolicyComponent policyComponent;
     private NodeService nodeService;
@@ -88,6 +90,7 @@ public class RoutingContentServiceTest extends TestCase
     @Override
     public void setUp() throws Exception
     {
+        transactionService = (TransactionService) ctx.getBean("transactionComponent");
         nodeService = (NodeService) ctx.getBean("dbNodeService");
         contentService = (ContentService) ctx.getBean(ServiceRegistry.CONTENT_SERVICE.getLocalName());
         this.policyComponent = (PolicyComponent) ctx.getBean("policyComponent");
@@ -148,7 +151,6 @@ public class RoutingContentServiceTest extends TestCase
     
     private UserTransaction getUserTransaction()
     {
-        TransactionService transactionService = (TransactionService) ctx.getBean("transactionComponent");
         return (UserTransaction) transactionService.getUserTransaction();
     }
     
@@ -552,58 +554,70 @@ public class RoutingContentServiceTest extends TestCase
         txn.rollback();
     }
     
+    /**
+     * Create several threads that will attempt to write to the same node property.
+     * The ContentWriter is handed to the thread, so this checks that the stream closure
+     * uses the transaction that called <code>close</code> and not the transaction that
+     * fetched the <code>ContentWriter</code>.
+     */
     public synchronized void testConcurrentWritesWithMultipleTxns() throws Exception
     {
-        // commit node so that threads can see node
-        txn.commit();
-        txn = null;
-        
-        UserTransaction txn = getUserTransaction();
-        txn.begin();
-        
         // ensure that there is no content to read on the node
         ContentReader reader = contentService.getReader(contentNodeRef, ContentModel.PROP_CONTENT);
         assertNull("Reader should not be available", reader);
 
-        ContentWriter threadWriter = contentService.getWriter(contentNodeRef, ContentModel.PROP_CONTENT, true);
+        // commit node so that threads can see node
+        txn.commit();
+        txn = null;
+        
         String threadContent = "Thread content";
-        WriteThread thread = new WriteThread(threadWriter, threadContent);
+        WriteThread[] writeThreads = new WriteThread[5];
-        // kick off thread
+        for (int i = 0; i < writeThreads.length; i++)
-        thread.start();
+        {
-        // wait for thread to get to its wait points
+            ContentWriter threadWriter = contentService.getWriter(contentNodeRef, ContentModel.PROP_CONTENT, true);
-        while (!thread.isWaiting())
+            writeThreads[i] = new WriteThread(threadWriter, threadContent);
+            // Kick each thread off
+            writeThreads[i].start();
+        }
+        
+        // Wait for all threads to be waiting
+        outer:
+        while (true)
+        {
+            // Wait for each thread to be in a transaction
+            for (int i = 0; i < writeThreads.length; i++)
+            {
+                if (!writeThreads[i].isWaiting())
+                {
+                    wait(10);
+                    continue outer;
+                }
+            }
+            // All threads were waiting
+            break outer;
+        }
+        
+        // Kick each thread into the stream close phase
+        for (int i = 0; i < writeThreads.length; i++)
+        {
+            synchronized(writeThreads[i])
+            {
+                writeThreads[i].notifyAll();
+            }
+        }
+        // Wait for the threads to complete (one way or another)
+        for (int i = 0; i < writeThreads.length; i++)
+        {
+            while (!writeThreads[i].isDone())
             {
                 wait(10);
             }
-        
-        // write to the content
-        ContentWriter writer = contentService.getWriter(contentNodeRef, ContentModel.PROP_CONTENT, true);
-        writer.putContent(SOME_CONTENT);
-        
-        // fire thread up again
-        synchronized(threadWriter)
-        {
-            threadWriter.notifyAll();
         }
-        // thread is released - but we have to wait for it to complete
-        while (!thread.isDone())
-        {
-            wait(10);
-        }
-
-        // the thread has finished and has committed its changes - check the visibility
-        reader = contentService.getReader(contentNodeRef, ContentModel.PROP_CONTENT);
-        assertNotNull("Reader should now be available", reader);
-        String checkContent = reader.getContentString();
-        assertEquals("Content check failed", SOME_CONTENT, checkContent);
-        
-        // rollback the txn
-        txn.rollback();
 
         // check content has taken on thread's content
         reader = contentService.getReader(contentNodeRef, ContentModel.PROP_CONTENT);
         assertNotNull("Reader should now be available", reader);
-        checkContent = reader.getContentString();
+        String checkContent = reader.getContentString();
         assertEquals("Content check failed", threadContent, checkContent);
     }
     
@@ -666,14 +680,15 @@ public class RoutingContentServiceTest extends TestCase
      * output stream before terminating.
      * <p>
      * When firing thread up, be sure to call <code>notify</code> on the
-     * writer in order to let the thread run to completion.
+     * Thread instance in order to let the thread run to completion.
      */
     private class WriteThread extends Thread
     {
         private ContentWriter writer;
         private String content;
-        private boolean isWaiting;
+        private volatile boolean isWaiting;
-        private boolean isDone;
+        private volatile boolean isDone;
+        private volatile Throwable error;
         
         public WriteThread(ContentWriter writer, String content)
         {
@@ -681,6 +696,7 @@ public class RoutingContentServiceTest extends TestCase
             this.content = content;
             isWaiting = false;
             isDone = false;
+            error = null;
         }
         
         public boolean isWaiting()
@@ -693,17 +709,29 @@ public class RoutingContentServiceTest extends TestCase
             return isDone;
         }
         
+        public Throwable getError()
+        {
+            return error;
+        }
+
         public void run()
         {
             authenticationComponent.setSystemUserAsCurrentUser();
             
-            isWaiting = false;
+            synchronized (this)
-            isDone = false;
+            {
-            UserTransaction txn = getUserTransaction();
+                isWaiting = true;
-            OutputStream os = writer.getContentOutputStream();
+                try { this.wait(); } catch (InterruptedException e) {};   // wait until notified
+            }
+
+            final OutputStream os = writer.getContentOutputStream();
+            // Callback to write to the content in a new transaction
+            RetryingTransactionCallback<Void> callback = new RetryingTransactionCallback<Void>()
+            {
+                public Void execute() throws Throwable
+                {
                     try
                     {
-                txn.begin();    // not testing transactions - this is not a safe pattern
                         // put the content
                         if (writer.getEncoding() == null)
                         {
@@ -713,20 +741,7 @@ public class RoutingContentServiceTest extends TestCase
                         {
                             os.write(content.getBytes(writer.getEncoding()));
                         }
-                synchronized (writer)
-                {
-                    isWaiting = true;
-                    writer.wait();   // wait until notified
-                }
                         os.close();
-                os = null;
-                txn.commit();
-            }
-            catch (Throwable e)
-            {
-                try {txn.rollback(); } catch (Exception ee) {}
-                e.printStackTrace();
-                throw new RuntimeException("Failed writing to output stream for writer: " + writer, e);
                     }
                     finally
                     {
@@ -734,6 +749,21 @@ public class RoutingContentServiceTest extends TestCase
                         {
                             try { os.close(); } catch (IOException e) {}
                         }
+                    }
+                    return null;
+                }
+            };
+            try
+            {
+                transactionService.getRetryingTransactionHelper().doInTransaction(callback);
+            }
+            catch (Throwable e)
+            {
+                e.printStackTrace();
+                error = e;
+            }
+            finally
+            {
                 isDone = true;
             }
         }

source/java/org/alfresco/repo/domain/contentdata/ibatis/ContentDataDAOImpl.java

@@ -159,6 +159,11 @@ public class ContentDataDAOImpl extends AbstractContentDataDAOImpl
             {
                 // Get the content urls
                 ContentDataEntity contentDataEntity = getContentDataEntity(id);
+                // This might be null as there is no constraint ensuring that the node points to a valid ContentData entity
+                if (contentDataEntity == null)
+                {
+                    continue;
+                }
                 // Only check the content URLs if one is present
                 String contentUrl = contentDataEntity.getContentUrl();
                 Long contentUrlId = contentDataEntity.getContentUrlId();

source/java/org/alfresco/repo/node/db/hibernate/HibernateNodeDaoServiceImpl.java

@@ -1348,7 +1348,9 @@ public class HibernateNodeDaoServiceImpl extends HibernateDaoSupport implements
         if (propertyDef != null && propertyDef.getDataType().getName().equals(DataTypeDefinition.CONTENT))
         {
             Set<Long> contentQNamesToRemoveIds = Collections.singleton(qnameId);
-            contentDataDAO.deleteContentDataForNode(node.getId(), contentQNamesToRemoveIds);
+            contentDataDAO.deleteContentDataForNode(
+                    node.getId(),
+                    contentQNamesToRemoveIds);
         }
 
         Map<PropertyMapKey, NodePropertyValue> persistableProperties = new HashMap<PropertyMapKey, NodePropertyValue>(3);