Revert "fix: properly close executor services and HTTP clients (closes #2)"

This reverts commit a34f94e.

Author: Stuart Hart

commit.bat

@@ -116,7 +116,7 @@ public void consume(String payload) {
         } catch (Exception e) {
             Log.errorf(e, "Failed to process position event: %s", payload);
             // SmallRye DLQ strategy will route this to position.dlq
-            throw new PositionProcessingException("Position processing failed", e);
+            throw new RuntimeException("Position processing failed", e);
         }
     }
 

geotrack-processing/src/main/java/com/geotrack/processing/consumer/PositionEventConsumer.java

@@ -7,8 +7,6 @@
 import com.geotrack.simulator.route.RouteReplayer;
 import io.quarkus.logging.Log;
 import jakarta.enterprise.context.Dependent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import jakarta.inject.Inject;
 import picocli.CommandLine.Command;
 import picocli.CommandLine.Option;
@@ -27,8 +25,6 @@
         description = "GeoTrack fleet simulation and GPX route replay")
 public class SimulatorCommand implements Runnable {
 
-    private static final Logger log = LoggerFactory.getLogger(SimulatorCommand.class);
-
     @Option(names = {"--fleet"}, description = "Run Newcastle demo fleet (4 vehicles)")
     boolean fleet;
 
@@ -78,7 +74,7 @@ private void runFleet() {
         var simulator = new FleetSimulator(speedMultiplier).withNewcastleFleet();
         var futures = simulator.simulate(position -> {
             if (dryRun) {
-                log.info("[{}] {}, {} | speed={} km/h heading={}°",
+                System.out.printf("[%s] %.6f, %.6f | speed=%.1f km/h heading=%.0f°%n",
                         position.assetId(), position.latitude(), position.longitude(),
                         position.speed(), position.heading());
             } else {
@@ -109,7 +105,7 @@ private void runGpxReplay() {
 
             new RouteReplayer(route, assetId, speedMultiplier).replay(position -> {
                 if (dryRun) {
-                    log.info("[{}] {}, {} | speed={} km/h",
+                    System.out.printf("[%s] %.6f, %.6f | speed=%.1f km/h%n",
                             position.assetId(), position.latitude(), position.longitude(), position.speed());
                 } else {
                     kafkaProducer.send(position);

geotrack-processing/src/main/java/com/geotrack/processing/consumer/PositionProcessingException.java

@@ -53,41 +53,27 @@ public void run() {
             return t;
         });
 
-        try {
-            // Graceful shutdown on SIGINT
-            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
-                Log.info("🛑 Shutting down OpenSky ingestor...");
-                scheduler.shutdown();
-                shutdownLatch.countDown();
-            }));
-
-            scheduler.scheduleAtFixedRate(() -> {
-                try {
-                    ingestor.poll();
-                } catch (Exception e) {
-                    Log.errorf(e, "Error during OpenSky poll cycle");
-                }
-            }, 0, intervalSeconds, TimeUnit.SECONDS);
+        // Graceful shutdown on SIGINT
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            Log.info("🛑 Shutting down OpenSky ingestor...");
+            scheduler.shutdown();
+            shutdownLatch.countDown();
+        }));
 
-            // Block until shutdown signal
+        scheduler.scheduleAtFixedRate(() -> {
             try {
-                shutdownLatch.await();
-                scheduler.awaitTermination(10, TimeUnit.SECONDS);
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-            }
-        } finally {
-            if (!scheduler.isShutdown()) {
-                scheduler.shutdown();
-            }
-            try {
-                if (!scheduler.awaitTermination(10, TimeUnit.SECONDS)) {
-                    scheduler.shutdownNow();
-                }
-            } catch (InterruptedException e) {
-                scheduler.shutdownNow();
-                Thread.currentThread().interrupt();
+                ingestor.poll();
+            } catch (Exception e) {
+                Log.errorf(e, "Error during OpenSky poll cycle");
             }
+        }, 0, intervalSeconds, TimeUnit.SECONDS);
+
+        // Block until shutdown signal
+        try {
+            shutdownLatch.await();
+            scheduler.awaitTermination(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
         }
 
         Log.info("✈ OpenSky ingestor stopped");

geotrack-simulator/src/main/java/com/geotrack/simulator/SimulatorCommand.java

@@ -11,7 +11,6 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
 /**
@@ -73,14 +72,6 @@ public List<Future<?>> simulate(Consumer<Position> positionConsumer) {
                 .toList());
 
         executor.shutdown();
-        try {
-            if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
-                executor.shutdownNow();
-            }
-        } catch (InterruptedException e) {
-            executor.shutdownNow();
-            Thread.currentThread().interrupt();
-        }
         return futures;
     }
 }

geotrack-simulator/src/main/java/com/geotrack/simulator/command/LiveIngestCommand.java

@@ -48,7 +48,6 @@ public class AISStreamIngestor {
     private final AtomicInteger totalReceived = new AtomicInteger();
     private final AtomicBoolean running = new AtomicBoolean(true);
     private final ScheduledExecutorService scheduler;
-    private HttpClient httpClient;
 
     public AISStreamIngestor(PositionProducer producer, String apiKey) {
         this.producer = producer;
@@ -75,18 +74,6 @@ public void start() {
     public void stop() {
         running.set(false);
         scheduler.shutdown();
-        try {
-            if (!scheduler.awaitTermination(10, TimeUnit.SECONDS)) {
-                scheduler.shutdownNow();
-            }
-        } catch (InterruptedException e) {
-            scheduler.shutdownNow();
-            Thread.currentThread().interrupt();
-        }
-        if (httpClient != null) {
-            httpClient.close();
-            httpClient = null;
-        }
     }
 
     private void connect() {
@@ -98,8 +85,8 @@ private void connect() {
                 {"Apikey":"%s","BoundingBoxes":[[[49.0,-12.0],[61.0,3.0]]],"FilterMessageTypes":["PositionReport"]}""",
                 apiKey);
 
-        this.httpClient = HttpClient.newHttpClient();
-        httpClient.newWebSocketBuilder()
+        HttpClient.newHttpClient()
+                .newWebSocketBuilder()
                 .buildAsync(URI.create(WS_URL), new WebSocket.Listener() {
 
                     private final StringBuilder buffer = new StringBuilder();

geotrack-simulator/src/main/java/com/geotrack/simulator/fleet/FleetSimulator.java

@@ -45,7 +45,7 @@ public List<RoutePoint> parse(InputStream input) {
 
             return points;
         } catch (Exception e) {
-            throw new GpxParseException("Failed to parse GPX: " + e.getMessage(), e);
+            throw new RuntimeException("Failed to parse GPX: " + e.getMessage(), e);
         }
     }