diff --git a/solr/contrib/dataimporthandler/CHANGES.txt b/solr/contrib/dataimporthandler/CHANGES.txt
index 663beb23ec9..310124cbedf 100644
--- a/solr/contrib/dataimporthandler/CHANGES.txt
+++ b/solr/contrib/dataimporthandler/CHANGES.txt
@@ -14,7 +14,8 @@ $Id$
 
 ==================  3.3.0-dev ==============
 
-(No Changes)
+* SOLR-2551: Check dataimport.properties for write access (if delta-import is supported
+  in DIH configuration) before starting an import (C S, shalin)
 
 ==================  3.2.0 ==================
 
diff --git a/solr/contrib/dataimporthandler/src/main/java/org/apache/solr/handler/dataimport/DataImporter.java b/solr/contrib/dataimporthandler/src/main/java/org/apache/solr/handler/dataimport/DataImporter.java
index 7c7bc3beb0d..85ad0930c5d 100644
--- a/solr/contrib/dataimporthandler/src/main/java/org/apache/solr/handler/dataimport/DataImporter.java
+++ b/solr/contrib/dataimporthandler/src/main/java/org/apache/solr/handler/dataimport/DataImporter.java
@@ -39,6 +39,7 @@ import org.apache.commons.io.IOUtils;
 
 import javax.xml.parsers.DocumentBuilder;
 import javax.xml.parsers.DocumentBuilderFactory;
+import java.io.File;
 import java.io.StringReader;
 import java.text.SimpleDateFormat;
 import java.util.*;
@@ -84,6 +85,8 @@ public class DataImporter {
 
   private final Map<String , Object> coreScopeSession;
 
+  private boolean isDeltaImportSupported = false;
+
   /**
    * Only for testing purposes
    */
@@ -112,7 +115,9 @@ public class DataImporter {
       initEntity(e, fields, false);
       verifyWithSchema(fields);
       identifyPk(e);
-    }    
+      if (e.allAttributes.containsKey(SqlEntityProcessor.DELTA_QUERY))
+        isDeltaImportSupported = true;
+    }
   }
 
   private void verifyWithSchema(Map<String, DataConfig.Field> fields) {
@@ -349,6 +354,7 @@ public class DataImporter {
 
     try {
       docBuilder = new DocBuilder(this, writer, requestParams);
+      checkWritablePersistFile(writer);
       docBuilder.execute();
       if (!requestParams.debug)
         cumulativeStatistics.add(docBuilder.importStatistics);
@@ -363,6 +369,15 @@ public class DataImporter {
 
   }
 
+  private void checkWritablePersistFile(SolrWriter writer) {
+    File persistFile = writer.getPersistFile();
+    boolean isWritable = persistFile.exists() ? persistFile.canWrite() : persistFile.getParentFile().canWrite();
+    if (isDeltaImportSupported && !isWritable) {
+      throw new DataImportHandlerException(SEVERE, persistFile.getAbsolutePath() +
+          " is not writable. Delta imports are supported by data config but will not work.");
+    }
+  }
+
   public void doDeltaImport(SolrWriter writer, RequestParams requestParams) {
     LOG.info("Starting Delta Import");
     setStatus(Status.RUNNING_DELTA_DUMP);
@@ -370,6 +385,7 @@ public class DataImporter {
     try {
       setIndexStartTime(new Date());
       docBuilder = new DocBuilder(this, writer, requestParams);
+      checkWritablePersistFile(writer);
       docBuilder.execute();
       if (!requestParams.debug)
         cumulativeStatistics.add(docBuilder.importStatistics);
diff --git a/solr/contrib/dataimporthandler/src/main/java/org/apache/solr/handler/dataimport/SolrWriter.java b/solr/contrib/dataimporthandler/src/main/java/org/apache/solr/handler/dataimport/SolrWriter.java
index f548316b4f4..e7bbb6c000f 100644
--- a/solr/contrib/dataimporthandler/src/main/java/org/apache/solr/handler/dataimport/SolrWriter.java
+++ b/solr/contrib/dataimporthandler/src/main/java/org/apache/solr/handler/dataimport/SolrWriter.java
@@ -99,13 +99,10 @@ public class SolrWriter {
 
     try {
       props.putAll(p);
-      String filePath = configDir;
-      if (configDir != null && !configDir.endsWith(File.separator))
-        filePath += File.separator;
-      filePath += persistFilename;
-      propOutput = new FileOutputStream(filePath);
+      File persistFile = getPersistFile();
+      propOutput = new FileOutputStream(persistFile);
       props.store(propOutput, null);
-      log.info("Wrote last indexed time to " + persistFilename);
+      log.info("Wrote last indexed time to " + persistFile.getAbsolutePath());
     } catch (FileNotFoundException e) {
       throw new DataImportHandlerException(DataImportHandlerException.SEVERE,
               "Unable to persist Index Start Time", e);
@@ -122,6 +119,14 @@ public class SolrWriter {
     }
   }
 
+  File getPersistFile() {
+    String filePath = configDir;
+    if (configDir != null && !configDir.endsWith(File.separator))
+      filePath += File.separator;
+    filePath += persistFilename;
+    return new File(filePath);
+  }
+
   void finish() {
     try {
       processor.finish();
diff --git a/solr/contrib/dataimporthandler/src/test/java/org/apache/solr/handler/dataimport/TestSqlEntityProcessorDelta.java b/solr/contrib/dataimporthandler/src/test/java/org/apache/solr/handler/dataimport/TestSqlEntityProcessorDelta.java
index 60f1ae5ecd0..4cddebaa2e2 100644
--- a/solr/contrib/dataimporthandler/src/test/java/org/apache/solr/handler/dataimport/TestSqlEntityProcessorDelta.java
+++ b/solr/contrib/dataimporthandler/src/test/java/org/apache/solr/handler/dataimport/TestSqlEntityProcessorDelta.java
@@ -20,6 +20,8 @@ import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
+import java.io.File;
+import java.io.FileOutputStream;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -92,7 +94,37 @@ public class TestSqlEntityProcessorDelta extends AbstractDataImportHandlerTestCa
   public void testCompositePk_FullImport() throws Exception {
     add1document();
   }
-  
+
+  @Test
+  @SuppressWarnings("unchecked")
+  public void testNonWritablePersistFile() throws Exception {
+    // See SOLR-2551
+    String configDir = h.getCore().getResourceLoader().getConfigDir();
+    String filePath = configDir;
+    if (configDir != null && !configDir.endsWith(File.separator))
+      filePath += File.separator;
+    filePath += "dataimport.properties";
+    File f = new File(filePath);
+    // execute the test only if we are able to set file to read only mode
+    if ((f.exists() || f.createNewFile()) && f.setReadOnly()) {
+      try {
+        List parentRow = new ArrayList();
+        parentRow.add(createMap("id", "1"));
+        MockDataSource.setIterator(FULLIMPORT_QUERY, parentRow.iterator());
+
+        List childRow = new ArrayList();
+        childRow.add(createMap("desc", "hello"));
+        MockDataSource.setIterator("select * from y where y.A='1'", childRow
+            .iterator());
+
+        runFullImport(dataConfig_delta);
+        assertQ(req("id:1"), "//*[@numFound='0']");
+      } finally {
+        f.setWritable(true);
+      }
+    }
+  }
+
   // WORKS
 
   @Test