do not silently ignore exceptions in PBF import (graphhopper#2270)

* OSM PBF import: do not silently ignore exceptions, fixes graphhopper#2269

* revert unrelated change

* minor name change

Author: karussell

core/src/main/java/com/graphhopper/reader/osm/OSMInputFile.java

@@ -46,12 +46,13 @@ public class OSMInputFile implements Sink, OSMInput {
     private final InputStream bis;
     private final BlockingQueue<ReaderElement> itemQueue;
     private final Queue<ReaderElement> itemBatch;
-    Thread pbfReaderThread;
     private boolean eof;
     // for xml parsing
-    private XMLStreamReader parser;
+    private XMLStreamReader xmlParser;
     // for pbf parsing
     private boolean binary = false;
+    private PbfReader pbfReader;
+    private Thread pbfReaderThread;
     private boolean hasIncomingData;
     private int workerThreads = -1;
     private OSMFileHeader fileheader;
@@ -138,17 +139,17 @@ private InputStream decode(File file) throws IOException {
     private void openXMLStream(InputStream in)
             throws XMLStreamException {
         XMLInputFactory factory = XMLInputFactory.newInstance();
-        parser = factory.createXMLStreamReader(in, "UTF-8");
+        xmlParser = factory.createXMLStreamReader(in, "UTF-8");
 
-        int event = parser.next();
-        if (event != XMLStreamConstants.START_ELEMENT || !parser.getLocalName().equalsIgnoreCase("osm")) {
+        int event = xmlParser.next();
+        if (event != XMLStreamConstants.START_ELEMENT || !xmlParser.getLocalName().equalsIgnoreCase("osm")) {
             throw new IllegalArgumentException("File is not a valid OSM stream");
         }
         // See https://wiki.openstreetmap.org/wiki/PBF_Format#Definition_of_the_OSMHeader_fileblock
-        String timestamp = parser.getAttributeValue(null, "osmosis_replication_timestamp");
+        String timestamp = xmlParser.getAttributeValue(null, "osmosis_replication_timestamp");
 
         if (timestamp == null)
-            timestamp = parser.getAttributeValue(null, "timestamp");
+            timestamp = xmlParser.getAttributeValue(null, "timestamp");
 
         if (timestamp != null) {
             try {
@@ -181,7 +182,7 @@ public ReaderElement getNext() throws XMLStreamException {
 
     private ReaderElement getNextXML() throws XMLStreamException {
 
-        int event = parser.next();
+        int event = xmlParser.next();
         if (fileheader != null) {
             ReaderElement copyfileheader = fileheader;
             fileheader = null;
@@ -190,32 +191,32 @@ private ReaderElement getNextXML() throws XMLStreamException {
 
         while (event != XMLStreamConstants.END_DOCUMENT) {
             if (event == XMLStreamConstants.START_ELEMENT) {
-                String idStr = parser.getAttributeValue(null, "id");
+                String idStr = xmlParser.getAttributeValue(null, "id");
                 if (idStr != null) {
-                    String name = parser.getLocalName();
+                    String name = xmlParser.getLocalName();
                     long id = 0;
                     switch (name.charAt(0)) {
                         case 'n':
                             // note vs. node
                             if ("node".equals(name)) {
                                 id = Long.parseLong(idStr);
-                                return OSMXMLHelper.createNode(id, parser);
+                                return OSMXMLHelper.createNode(id, xmlParser);
                             }
                             break;
 
                         case 'w': {
                             id = Long.parseLong(idStr);
-                            return OSMXMLHelper.createWay(id, parser);
+                            return OSMXMLHelper.createWay(id, xmlParser);
                         }
                         case 'r':
                             id = Long.parseLong(idStr);
-                            return OSMXMLHelper.createRelation(id, parser);
+                            return OSMXMLHelper.createRelation(id, xmlParser);
                     }
                 }
             }
-            event = parser.next();
+            event = xmlParser.next();
         }
-        parser.close();
+        xmlParser.close();
         return null;
     }
 
@@ -226,8 +227,10 @@ public boolean isEOF() {
     @Override
     public void close() throws IOException {
         try {
-            if (!binary)
-                parser.close();
+            if (binary)
+                pbfReader.close();
+            else
+                xmlParser.close();
         } catch (XMLStreamException ex) {
             throw new IOException(ex);
         } finally {
@@ -244,8 +247,8 @@ private void openPBFReader(InputStream stream) {
         if (workerThreads <= 0)
             workerThreads = 1;
 
-        PbfReader reader = new PbfReader(stream, this, workerThreads);
-        pbfReaderThread = new Thread(reader, "PBF Reader");
+        pbfReader = new PbfReader(stream, this, workerThreads);
+        pbfReaderThread = new Thread(pbfReader, "PBF Reader");
         pbfReaderThread.start();
     }
 
@@ -273,7 +276,7 @@ private ReaderElement getNextPBF() {
             if (!hasIncomingData && itemQueue.isEmpty()) {
                 return null; // signal EOF
             }
-            
+
             if (itemQueue.drainTo(itemBatch, MAX_BATCH_SIZE) == 0) {
                 try {
                     ReaderElement element = itemQueue.poll(100, TimeUnit.MILLISECONDS);
@@ -286,7 +289,7 @@ private ReaderElement getNextPBF() {
                 }
             }
         }
-        
+
         return itemBatch.poll();
     }
 }

core/src/main/java/com/graphhopper/reader/osm/pbf/PbfBlobDecoder.java

@@ -356,6 +356,7 @@ public void run() {
             listener.complete(decodedEntities);
 
         } catch (RuntimeException e) {
+            // exception is properly rethrown in PbfDecoder.sendResultsToSink
             listener.error(e);
         }
     }

core/src/main/java/com/graphhopper/reader/osm/pbf/PbfDecoder.java

@@ -3,11 +3,9 @@
 
 import com.graphhopper.reader.ReaderElement;
 
-import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Queue;
-import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
@@ -20,7 +18,7 @@
  *
  * @author Brett Henderson
  */
-public class PbfDecoder implements Runnable {
+public class PbfDecoder {
     private final PbfStreamSplitter streamSplitter;
     private final ExecutorService executorService;
     private final int maxPendingBlobs;
@@ -155,7 +153,6 @@ public void complete(List<ReaderElement> decodedEntities) {
         sendResultsToSink(0);
     }
 
-    @Override
     public void run() {
         lock.lock();
         try {

core/src/main/java/com/graphhopper/reader/osm/pbf/PbfReader.java

@@ -13,6 +13,7 @@
  * @author Brett Henderson
  */
 public class PbfReader implements Runnable {
+    private Throwable throwable;
     private InputStream inputStream;
     private Sink sink;
     private int workers;
@@ -33,10 +34,10 @@ public PbfReader(InputStream in, Sink sink, int workers) {
     @Override
     public void run() {
         ExecutorService executorService = Executors.newFixedThreadPool(workers);
-        try {
-            // Create a stream splitter to break the PBF stream into blobs.
-            PbfStreamSplitter streamSplitter = new PbfStreamSplitter(new DataInputStream(inputStream));
+        // Create a stream splitter to break the PBF stream into blobs.
+        PbfStreamSplitter streamSplitter = new PbfStreamSplitter(new DataInputStream(inputStream));
 
+        try {
             // Process all blobs of data in the stream using threads from the
             // executor service. We allow the decoder to issue an extra blob
             // than there are workers to ensure there is another blob
@@ -46,11 +47,18 @@ public void run() {
             PbfDecoder pbfDecoder = new PbfDecoder(streamSplitter, executorService, workers + 1, sink);
             pbfDecoder.run();
 
-        } catch (Exception e) {
-            throw new RuntimeException("Unable to read PBF file.", e);
+        } catch (Throwable t) {
+            // properly propagate exception inside Thread, #2269
+            throwable = t;
         } finally {
             sink.complete();
             executorService.shutdownNow();
+            streamSplitter.release();
         }
     }
+
+    public void close() {
+        if (throwable != null)
+            throw new RuntimeException("Unable to read PBF file.", throwable);
+    }
 }

core/src/main/java/com/graphhopper/reader/osm/pbf/PbfStreamSplitter.java

@@ -110,7 +110,7 @@ public void release() {
             try {
                 dis.close();
             } catch (IOException e) {
-                log.log(Level.SEVERE, "Unable to close PBF stream.", e);
+                throw new RuntimeException(e);
             }
         }
         dis = null;