From 4f25ea267106ada5f9e0399498f4280a05caf045 Mon Sep 17 00:00:00 2001 From: Jeremy Fee Date: Mon, 28 Dec 2020 11:13:05 -0700 Subject: [PATCH 1/6] Make json product geometry handle nulls --- .../earthquake/product/io/JsonProduct.java | 20 +++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/src/main/java/gov/usgs/earthquake/product/io/JsonProduct.java b/src/main/java/gov/usgs/earthquake/product/io/JsonProduct.java index 782f4f3e..20d3676f 100644 --- a/src/main/java/gov/usgs/earthquake/product/io/JsonProduct.java +++ b/src/main/java/gov/usgs/earthquake/product/io/JsonProduct.java @@ -161,10 +161,26 @@ public JsonObjectBuilder getGeometryJson(final Product product) throws Exception final BigDecimal latitude = product.getLatitude(); final BigDecimal longitude = product.getLongitude(); final BigDecimal depth = product.getDepth(); - if (latitude != null || longitude != null) { + if (latitude != null || longitude != null || depth != null) { + final JsonArrayBuilder coordinates = Json.createArrayBuilder(); + if (latitude != null) { + coordinates.add(latitude); + } else { + coordinates.addNull(); + } + if (longitude != null) { + coordinates.add(longitude); + } else { + coordinates.addNull(); + } + if (depth != null) { + coordinates.add(depth); + } else { + coordinates.addNull(); + } return Json.createObjectBuilder() .add("type", "Point") - .add("coordinates", Json.createArrayBuilder().add(longitude).add(latitude).add(depth)); + .add("coordinates", coordinates); } return null; } From 360a99aeb139a83da31fe0b27ed27c3c21045452 Mon Sep 17 00:00:00 2001 From: Jeremy Fee Date: Mon, 28 Dec 2020 11:13:53 -0700 Subject: [PATCH 2/6] Increase timeout for json notification index --- .../gov/usgs/earthquake/aws/JsonNotificationIndex.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/main/java/gov/usgs/earthquake/aws/JsonNotificationIndex.java b/src/main/java/gov/usgs/earthquake/aws/JsonNotificationIndex.java index 6eef3bb0..2d6668c3 100644 --- a/src/main/java/gov/usgs/earthquake/aws/JsonNotificationIndex.java +++ b/src/main/java/gov/usgs/earthquake/aws/JsonNotificationIndex.java @@ -345,7 +345,7 @@ public synchronized List findNotifications( beginTransaction(); try (final PreparedStatement statement = getConnection().prepareStatement(sql)) { try { - statement.setQueryTimeout(1000); + statement.setQueryTimeout(1800); // set parameters for (int i = 0, len=values.size(); i < len; i++) { @@ -429,7 +429,7 @@ public synchronized List findNotifications( beginTransaction(); try (final PreparedStatement statement = getConnection().prepareStatement(sql)) { try { - statement.setQueryTimeout(1000); + statement.setQueryTimeout(1800); // set parameters for (int i = 0, len=values.size(); i < len; i++) { @@ -465,7 +465,7 @@ public synchronized List findExpiredNotifications() throws Excepti beginTransaction(); try (final PreparedStatement statement = getConnection().prepareStatement(sql)) { try { - statement.setQueryTimeout(1000); + statement.setQueryTimeout(1800); // set parameters statement.setString(1, Instant.now().toString()); @@ -562,7 +562,7 @@ public synchronized List getMissingNotifications( beginTransaction(); try (final PreparedStatement statement = getConnection().prepareStatement(sql)) { try { - statement.setQueryTimeout(1000); + statement.setQueryTimeout(1800); // execute and commit if successful final List notifications = getNotifications(statement); commitTransaction(); From 866aa088cfbb7a9794fe1c38b91bd3f676292baa Mon Sep 17 00:00:00 2001 From: Jeremy Fee Date: Mon, 28 Dec 2020 11:14:20 -0700 Subject: [PATCH 3/6] Use transactions for readProductIndex to reconnect when needed --- .../gov/usgs/earthquake/indexer/Indexer.java | 23 ++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/src/main/java/gov/usgs/earthquake/indexer/Indexer.java b/src/main/java/gov/usgs/earthquake/indexer/Indexer.java index 3d4d4736..3c4befb9 100644 --- a/src/main/java/gov/usgs/earthquake/indexer/Indexer.java +++ b/src/main/java/gov/usgs/earthquake/indexer/Indexer.java @@ -420,11 +420,28 @@ protected boolean hasProductBeenIndexed(final ProductId id) { if (readProductIndex == productIndex) { // synchronize on this if read and product index are same synchronized (indexProductSync) { - return readProductIndex.hasProduct(id); + readProductIndex.beginTransaction(); + try { + boolean hasProduct = readProductIndex.hasProduct(id); + readProductIndex.commitTransaction(); + return hasProduct; + } catch (Exception e) { + readProductIndex.rollbackTransaction(); + } } } else { - // otherwise readProductIndex provides own synchronization - return readProductIndex.hasProduct(id); + // otherwise synchronize on readProductIndex + // transaction reconnects if needed + synchronized (readProductIndex) { + readProductIndex.beginTransaction(); + try { + boolean hasProduct = readProductIndex.hasProduct(id); + readProductIndex.commitTransaction(); + return hasProduct; + } catch (Exception e) { + readProductIndex.rollbackTransaction(); + } + } } } catch (Exception wtf) { LOGGER.log(Level.WARNING, "[" + getName() From 292ac2a33f49d1e2257c7a4d1fae068f06e74968 Mon Sep 17 00:00:00 2001 From: Jeremy Fee Date: Mon, 4 Jan 2021 10:02:13 -0700 Subject: [PATCH 4/6] Make initialCatchUpAge configurable --- docs/userguide/configuration.html | 13 ++++++++++++ .../earthquake/aws/AwsProductReceiver.java | 20 ++++++++++++++++--- 2 files changed, 30 insertions(+), 3 deletions(-) diff --git a/docs/userguide/configuration.html b/docs/userguide/configuration.html index cff04088..fee84505 100644 --- a/docs/userguide/configuration.html +++ b/docs/userguide/configuration.html @@ -604,11 +604,24 @@

Receiver Types

trackingFileName
File where current position in hub notifications is tracked.
+
trackingIndex
+
+ Custom tracking index. + (by default created is a sqlite database at trackingFileName) +
+
connectAttempts
(Optional, default 5) Number to attempts to connect before pausing
connectTimeout
(Optional, default 1000ms) timeout for each connect attempt.
+ +
initialCatchUpAge
+
+ (Optional, default 7.0 days) how far back in time to start when + connecting to server. subsequent attempts start at current position, + which is tracked. +
diff --git a/src/main/java/gov/usgs/earthquake/aws/AwsProductReceiver.java b/src/main/java/gov/usgs/earthquake/aws/AwsProductReceiver.java index f19158ce..106fa91c 100644 --- a/src/main/java/gov/usgs/earthquake/aws/AwsProductReceiver.java +++ b/src/main/java/gov/usgs/earthquake/aws/AwsProductReceiver.java @@ -40,10 +40,12 @@ public class AwsProductReceiver extends DefaultNotificationReceiver implements R public static final String TRACKING_FILE_NAME_PROPERTY = "trackingFileName"; public static final String CONNECT_ATTEMPTS_PROPERTY = "connectAttempts"; public static final String CONNECT_TIMEOUT_PROPERTY = "connectTimeout"; + public static final String INITIAL_CATCHUP_AGE_PROPERTY = "initialCatchUpAge"; public static final String DEFAULT_TRACKING_FILE_NAME = "data/AwsReceiver.json"; public static final String DEFAULT_CONNECT_ATTEMPTS = "5"; public static final String DEFAULT_CONNECT_TIMEOUT = "1000"; + public static final String DEFAULT_INITIAL_CATCHUP_AGE = "7.0"; private URI uri; private String trackingFileName; @@ -58,6 +60,10 @@ public class AwsProductReceiver extends DefaultNotificationReceiver implements R /* µs timestamp of last message that has been processed */ protected Instant createdAfter = null; + + /** How far back to check when first connecting. */ + protected double initialCatchUpAge = Double.valueOf(DEFAULT_INITIAL_CATCHUP_AGE); + /* last broadcast message that has been processed (used for catch up) */ protected JsonNotification lastBroadcast = null; protected Long lastBroadcastId = null; @@ -80,8 +86,12 @@ public void configure(Config config) throws Exception { super.configure(config); uri = new URI(config.getProperty(URI_PROPERTY)); - attempts = Integer.parseInt(config.getProperty(CONNECT_ATTEMPTS_PROPERTY, DEFAULT_CONNECT_ATTEMPTS)); - timeout = Long.parseLong(config.getProperty(CONNECT_TIMEOUT_PROPERTY, DEFAULT_CONNECT_TIMEOUT)); + attempts = Integer.parseInt( + config.getProperty(CONNECT_ATTEMPTS_PROPERTY, DEFAULT_CONNECT_ATTEMPTS)); + timeout = Long.parseLong( + config.getProperty(CONNECT_TIMEOUT_PROPERTY, DEFAULT_CONNECT_TIMEOUT)); + initialCatchUpAge = Double.valueOf( + config.getProperty(INITIAL_CATCHUP_AGE_PROPERTY, DEFAULT_INITIAL_CATCHUP_AGE)); final String trackingIndexName = config.getProperty(TRACKING_INDEX_PROPERTY); if (trackingIndexName != null) { @@ -338,7 +348,8 @@ public void run() { protected void sendProductsCreatedAfter() throws IOException { // set default for created after if (this.createdAfter == null) { - this.createdAfter = Instant.now().minusSeconds(7 * 86400); + this.createdAfter = Instant.now().minusSeconds( + Math.round(initialCatchUpAge * 86400)); } String request = Json.createObjectBuilder() .add("action", "products_created_after") @@ -393,6 +404,9 @@ protected void stopCatchUp() { * Stop background thread for catch up process. */ protected void stopCatchUpThread() { + if (catchUpThread == null) { + return; + } // stop catch up thread try { synchronized (catchUpSync) { From 900261606663588f8048a045b6c63307d1e5a2e1 Mon Sep 17 00:00:00 2001 From: Jeremy Fee Date: Mon, 28 Dec 2020 11:24:35 -0700 Subject: [PATCH 5/6] Update version to 2.7.5 --- code.json | 4 ++-- .../java/gov/usgs/earthquake/distribution/ProductClient.java | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/code.json b/code.json index 6cfe0366..2448c29a 100644 --- a/code.json +++ b/code.json @@ -3,7 +3,7 @@ "name": "Product Distribution Layer", "organization": "U.S. Geological Survey", "description": "Distribution system used for derived earthquake information", - "version": "v2.7.4", + "version": "v2.7.5", "status": "Production", "permissions": { "usageType": "openSource", @@ -27,7 +27,7 @@ "email": "jmfee@usgs.gov" }, "date": { - "metadataLastUpdated": "2020-12-23" + "metadataLastUpdated": "2020-12-28" } } ] diff --git a/src/main/java/gov/usgs/earthquake/distribution/ProductClient.java b/src/main/java/gov/usgs/earthquake/distribution/ProductClient.java index ade3f17a..40ed0e5c 100644 --- a/src/main/java/gov/usgs/earthquake/distribution/ProductClient.java +++ b/src/main/java/gov/usgs/earthquake/distribution/ProductClient.java @@ -64,7 +64,7 @@ public class ProductClient extends DefaultConfigurable implements ProductClientMBean, Bootstrappable { /** The "release" version number. */ - public static final String RELEASE_VERSION = "Version 2.7.4 2020-12-23"; + public static final String RELEASE_VERSION = "Version 2.7.5 2020-12-28"; /** Property name used on products for current RELEASE_VERSION. */ public static final String PDL_CLIENT_VERSION_PROPERTY = "pdl-client-version"; From 366d8c1cdcdd4ab0e5dcad5359665d41c600e4e2 Mon Sep 17 00:00:00 2001 From: Jeremy Fee Date: Mon, 4 Jan 2021 10:05:20 -0700 Subject: [PATCH 6/6] Update 2.7.5 date to 2021-01-04 --- code.json | 2 +- .../java/gov/usgs/earthquake/distribution/ProductClient.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/code.json b/code.json index 2448c29a..8a8e3133 100644 --- a/code.json +++ b/code.json @@ -27,7 +27,7 @@ "email": "jmfee@usgs.gov" }, "date": { - "metadataLastUpdated": "2020-12-28" + "metadataLastUpdated": "2021-01-04" } } ] diff --git a/src/main/java/gov/usgs/earthquake/distribution/ProductClient.java b/src/main/java/gov/usgs/earthquake/distribution/ProductClient.java index 40ed0e5c..a1df371f 100644 --- a/src/main/java/gov/usgs/earthquake/distribution/ProductClient.java +++ b/src/main/java/gov/usgs/earthquake/distribution/ProductClient.java @@ -64,7 +64,7 @@ public class ProductClient extends DefaultConfigurable implements ProductClientMBean, Bootstrappable { /** The "release" version number. */ - public static final String RELEASE_VERSION = "Version 2.7.5 2020-12-28"; + public static final String RELEASE_VERSION = "Version 2.7.5 2021-01-04"; /** Property name used on products for current RELEASE_VERSION. */ public static final String PDL_CLIENT_VERSION_PROPERTY = "pdl-client-version";