From d4a30447c42f572feeffd04c207e9f6c3417350e Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Sat, 7 Nov 2026 06:37:30 +0000 Subject: [PATCH 01/16] Test that usage sharing pauses sending after failed sends Usage sharing in Java has no back-off. Every batch is sent while the service is failing, and is lost. These tests expect the back-off .NET is being given, at its default settings of ten failed sends within 100 seconds and a first pause of two seconds: - after ten failed sends nothing is sent during the pause, and the evidence processed during it is all sent once the pause ends and the service answers, with no more requests processed; - closing while paused returns at once and sends nothing. They fail until the back-off is added. --- .../ShareUsageElementBackOffTests.java | 265 ++++++++++++++++++ 1 file changed, 265 insertions(+) create mode 100644 pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java diff --git a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java new file mode 100644 index 00000000..da10ee1e --- /dev/null +++ b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java @@ -0,0 +1,265 @@ +/* ********************************************************************* + * This Original Work is copyright of 51 Degrees Mobile Experts Limited. + * Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, + * Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. + * + * This Original Work is licensed under the European Union Public Licence + * (EUPL) v.1.2 and is subject to its terms as set out below. + * + * If a copy of the EUPL was not distributed with this file, You can obtain + * one at https://opensource.org/licenses/EUPL-1.2. + * + * The 'Compatible Licences' set out in the Appendix to the EUPL (as may be + * amended by the European Commission) shall be deemed incompatible for + * the purposes of the Work and the provisions of the compatibility + * clause in Article 5 of the EUPL shall not apply. + * + * If using the Work as, or as part of, a network application, by + * including the attribution notice(s) required under Article 5 of the EUPL + * in the end user terms of the application under an appropriate heading, + * such notice(s) shall fulfill the requirements of that article. + * ********************************************************************* */ + +package fiftyone.pipeline.engines.fiftyone.flowelements; + +import fiftyone.pipeline.core.data.FlowData; +import fiftyone.pipeline.core.flowelements.Pipeline; +import fiftyone.pipeline.core.flowelements.PipelineBuilder; +import fiftyone.pipeline.engines.services.DataUploader; +import fiftyone.pipeline.engines.trackers.Tracker; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; + +import java.io.ByteArrayOutputStream; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.function.BooleanSupplier; +import java.util.function.Supplier; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import static fiftyone.pipeline.core.Constants.EVIDENCE_CLIENTIP_KEY; +import static fiftyone.pipeline.engines.Constants.DEFAULT_SESSION_COOKIE_NAME; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; + +/** + * Usage sharing pauses sending after repeated failed sends, keeps queueing + * evidence while paused, and drains the queue once a send succeeds. These + * tests use the default settings, which pause sending for two seconds after + * ten sends fail within 100 seconds. + */ +public class ShareUsageElementBackOffTests { + + /** + * The longest a test waits for something it expects to happen. + */ + private static final long WAIT_LIMIT_MILLIS = 30000; + + private static final int DEFAULT_FAILURES = 10; + + private static final long DEFAULT_PAUSE_MILLIS = 2000; + + private static final Pattern CLIENT_IP = + Pattern.compile("([^<]*)"); + + /** + * Answers each send with the status the test sets, and records when each + * send was answered and the client IPs of the sends it accepted. + */ + private static class RecordingUploader implements DataUploader { + private final long start = System.nanoTime(); + private final List times = new ArrayList<>(); + private final List accepted = new ArrayList<>(); + private volatile int code = 500; + private ByteArrayOutputStream body; + + @Override + public synchronized OutputStream getOutputStream() { + body = new ByteArrayOutputStream(); + return body; + } + + @Override + public synchronized int getResponseCode() { + int answer = code; + times.add(now()); + if (answer == 200) { + Matcher matcher = CLIENT_IP.matcher( + new String(body.toByteArray(), StandardCharsets.UTF_8)); + while (matcher.find()) { + accepted.add(matcher.group(1)); + } + } + return answer; + } + + long now() { + return (System.nanoTime() - start) / 1000000; + } + + synchronized int count() { + return times.size(); + } + + synchronized List times() { + return new ArrayList<>(times); + } + + synchronized List accepted() { + return new ArrayList<>(accepted); + } + } + + /** + * Shares every request. + */ + private static class ShareAll implements Tracker { + @Override + public boolean track(FlowData flowData) { + return true; + } + } + + private Pipeline pipeline; + private RecordingUploader uploader; + private ShareUsageElement element; + private int requests; + + /** + * Each entry is sent on its own, as the minimum entries per message is + * one. + */ + @BeforeEach + public void init() throws Exception { + pipeline = new PipelineBuilder().build(); + uploader = new RecordingUploader(); + requests = 0; + element = new ShareUsageElement( + LoggerFactory.getLogger(ShareUsageElement.class), + 1, + 1, + 100, + 100, + 100, + 1, + true, + "http://localhost/usage", + new ArrayList<>(), + new ArrayList<>(), + new ArrayList<>(), + DEFAULT_SESSION_COOKIE_NAME, + new ShareAll()); + element.addPipeline(pipeline); + element.dataUploader = uploader; + } + + @AfterEach + public void cleanUp() throws Exception { + element.close(); + pipeline.close(); + } + + /** + * Processes a request whose evidence no other request has, and returns + * its client IP. + */ + private String process() throws Exception { + int number = requests++; + String ip = "10.0." + (number / 250) + "." + (number % 250 + 1); + Map evidence = new HashMap<>(); + evidence.put(EVIDENCE_CLIENTIP_KEY, ip); + try (FlowData data = pipeline.createFlowData()) { + data.addEvidence(evidence); + element.process(data); + } + return ip; + } + + private static void waitFor( + BooleanSupplier condition, + Supplier message) throws InterruptedException { + long end = System.currentTimeMillis() + WAIT_LIMIT_MILLIS; + while (condition.getAsBoolean() == false) { + if (System.currentTimeMillis() > end) { + fail(message.get()); + } + Thread.sleep(20); + } + } + + /** + * Processes requests one at a time, waiting for each to be sent, so each + * is a separate failed send. + */ + private void failSends(int count) throws Exception { + for (int i = 1; i <= count; i++) { + process(); + final int expected = i; + waitFor( + () -> uploader.count() == expected && + element.isRunning() == false, + () -> "Send " + expected + " was not made."); + } + } + + /** + * Ten failed sends pause sending. Nothing is sent during the pause, the + * evidence processed during it is queued, and all of it is sent once the + * pause ends and the service answers, with no more requests processed to + * start the sending. + */ + @Test + public void ShareUsageElement_BackOff_PausesThenDrainsQueue() + throws Exception { + failSends(DEFAULT_FAILURES); + long pausedAt = uploader.now(); + + List queued = new ArrayList<>(); + for (int i = 0; i < 5; i++) { + queued.add(process()); + } + Thread.sleep(700); + assertEquals(DEFAULT_FAILURES, uploader.count(), + "Sends were made during the pause that ten failed sends should " + + "have started."); + + uploader.code = 200; + waitFor( + () -> uploader.accepted().containsAll(queued), + () -> "The evidence processed during the pause was not all " + + "sent once the service answered. Sent: " + uploader.accepted()); + long waited = uploader.times().get(DEFAULT_FAILURES) - pausedAt; + assertTrue(waited >= DEFAULT_PAUSE_MILLIS - 500, + "Sending resumed " + waited + " ms after the pause started, " + + "before the two second pause ended."); + } + + /** + * Closing while paused returns at once rather than waiting for the pause + * to end, and sends nothing. + */ + @Test + public void ShareUsageElement_BackOff_CloseWhilePausedDoesNotWaitOrSend() + throws Exception { + failSends(DEFAULT_FAILURES); + process(); + process(); + + long start = System.currentTimeMillis(); + element.close(); + long took = System.currentTimeMillis() - start; + + assertTrue(took < 1500, + "Closing took " + took + " ms, as it waited for the pause."); + assertEquals(DEFAULT_FAILURES, uploader.count(), + "A send was made during the pause or when closing."); + } +} From b26878f087a81fa1c2afaedd83c498a526e84bbf Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Sat, 7 Nov 2026 06:46:39 +0000 Subject: [PATCH 02/16] Pause usage sharing sends after repeated failures Java now has the same back-off as .NET, with the same five settings and defaults. - After setFailuresToEnterRecovery sends fail within setFailuresWindowSeconds, sending pauses for setInitialRecoveryDelaySeconds. A send fails when it throws or the answer is not 200. - When a pause ends one send is tried. If it fails the delay is multiplied by setRecoveryMultiplier, up to setMaxRecoveryDelaySeconds. If it succeeds the back-off ends. - Processing is not affected. Evidence is queued during a pause and new evidence is dropped only when the queue is full. - The consumer waits out the pause itself, on a latch that closing releases, and then sends until fewer than the minimum entries per message are queued, so the queue drains when a pause ends without more requests being processed. - Closing ends a pause at once and sends nothing while paused, logging how many entries were not sent. ShareUsageBackOff holds the rules and takes the time from the caller, so its tests set the time. The consumer calls the new protected waitWhilePaused before each message and recordSendResult after it. New constructors take the five settings, and the existing ones use the defaults. --- .../fiftyone/flowelements/Constants.java | 39 ++++ .../flowelements/ShareUsageBackOff.java | 149 ++++++++++++++ .../fiftyone/flowelements/ShareUsageBase.java | 155 +++++++++++++++ .../flowelements/ShareUsageBuilder.java | 7 +- .../flowelements/ShareUsageBuilderBase.java | 91 +++++++++ .../flowelements/ShareUsageElement.java | 111 ++++++++++- .../flowelements/ShareUsageBackOffTests.java | 184 ++++++++++++++++++ .../ShareUsageElementBackOffTests.java | 93 +++++++++ 8 files changed, 825 insertions(+), 4 deletions(-) create mode 100644 pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java create mode 100644 pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOffTests.java diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/Constants.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/Constants.java index f53a7bf2..f2e536e8 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/Constants.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/Constants.java @@ -104,4 +104,43 @@ public class Constants { */ static final String EVIDENCE_HTTPHEADER_COOKIE_SUFFIX = "cookie"; + /** + * The default number of failed sends within the failures window that + * pauses sending. + */ + public static final int SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_DEFAULT = 10; + + /** + * The fewest failed sends that can be set to pause sending. + */ + public static final int SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MIN = 1; + + /** + * The most failed sends that can be set to pause sending. + */ + public static final int SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MAX = 100; + + /** + * The default time in seconds over which failed sends are counted. + */ + public static final int SHARE_USAGE_FAILURES_WINDOW_SECONDS_DEFAULT = 100; + + /** + * The default length in seconds of the first pause after failed sends. + */ + public static final double + SHARE_USAGE_EXPONENTIAL_BACKOFF_INITIAL_DELAY_SECONDS_DEFAULT = 2.0; + + /** + * The default longest pause in seconds. + */ + public static final double + SHARE_USAGE_EXPONENTIAL_BACKOFF_MAX_DELAY_SECONDS_DEFAULT = 300.0; + + /** + * The default multiplier applied to the pause each time the send tried + * when a pause ends fails. + */ + public static final double + SHARE_USAGE_EXPONENTIAL_BACKOFF_MULTIPLIER_DEFAULT = 2.0; } diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java new file mode 100644 index 00000000..3cf13cde --- /dev/null +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java @@ -0,0 +1,149 @@ +/* ********************************************************************* + * This Original Work is copyright of 51 Degrees Mobile Experts Limited. + * Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, + * Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. + * + * This Original Work is licensed under the European Union Public Licence + * (EUPL) v.1.2 and is subject to its terms as set out below. + * + * If a copy of the EUPL was not distributed with this file, You can obtain + * one at https://opensource.org/licenses/EUPL-1.2. + * + * The 'Compatible Licences' set out in the Appendix to the EUPL (as may be + * amended by the European Commission) shall be deemed incompatible for + * the purposes of the Work and the provisions of the compatibility + * clause in Article 5 of the EUPL shall not apply. + * + * If using the Work as, or as part of, a network application, by + * including the attribution notice(s) required under Article 5 of the EUPL + * in the end user terms of the application under an appropriate heading, + * such notice(s) shall fulfill the requirements of that article. + * ********************************************************************* */ + +package fiftyone.pipeline.engines.fiftyone.flowelements; + +import java.util.ArrayDeque; +import java.util.Deque; + +/** + * Decides when usage sharing may send. Sending pauses for the initial delay + * once the set number of sends fail within the window. A failed send after a + * pause multiplies the delay, up to the maximum, and a successful send ends + * the back-off. Only sending pauses, so evidence is still queued while + * paused. Each method takes the time from the caller, in milliseconds. + */ +class ShareUsageBackOff { + + private final int failuresToEnterRecovery; + private final long failuresWindowMillis; + private final double initialDelaySeconds; + private final double maxDelaySeconds; + private final double multiplier; + + /** + * When each recent failed send happened, oldest first. Empty while + * backing off. + */ + private final Deque failures = new ArrayDeque<>(); + + /** + * The length of the current pause in seconds, or zero when not backing + * off. + */ + private double delaySeconds = 0; + + /** + * When sending may start again. + */ + private long resumeAt = Long.MIN_VALUE; + + ShareUsageBackOff( + int failuresToEnterRecovery, + int failuresWindowSeconds, + double initialDelaySeconds, + double maxDelaySeconds, + double multiplier) { + if (failuresToEnterRecovery < 1) { + throw new IllegalArgumentException( + "failuresToEnterRecovery must be 1 or more."); + } + if (failuresWindowSeconds < 1) { + throw new IllegalArgumentException( + "failuresWindowSeconds must be 1 or more."); + } + if ((initialDelaySeconds > 0) == false) { + throw new IllegalArgumentException( + "initialDelaySeconds must be more than 0."); + } + if ((maxDelaySeconds > 0) == false) { + throw new IllegalArgumentException( + "maxDelaySeconds must be more than 0."); + } + if ((multiplier >= 1) == false) { + throw new IllegalArgumentException( + "multiplier must be 1 or more."); + } + this.failuresToEnterRecovery = failuresToEnterRecovery; + this.failuresWindowMillis = failuresWindowSeconds * 1000L; + this.initialDelaySeconds = initialDelaySeconds; + this.maxDelaySeconds = maxDelaySeconds; + this.multiplier = multiplier; + } + + /** + * @return true from the start of a pause until a send succeeds + */ + synchronized boolean isBackingOff() { + return delaySeconds > 0; + } + + /** + * @param now the time now in milliseconds + * @return the milliseconds left before sending may start again, or zero + * if it may start now + */ + synchronized long remaining(long now) { + return resumeAt > now ? resumeAt - now : 0; + } + + /** + * Records a failed send. + * @param now the time now in milliseconds + * @return the new pause in milliseconds if this failure started or + * lengthened one, otherwise zero + */ + synchronized long failed(long now) { + if (delaySeconds > 0) { + // The send tried when the last pause ended failed. + delaySeconds = Math.min( + delaySeconds * multiplier, + maxDelaySeconds); + } else { + failures.addLast(now); + while (failures.peekFirst() < now - failuresWindowMillis) { + failures.removeFirst(); + } + if (failures.size() < failuresToEnterRecovery) { + return 0; + } + failures.clear(); + delaySeconds = Math.min(initialDelaySeconds, maxDelaySeconds); + } + long pause = Math.max(1, Math.round(delaySeconds * 1000)); + resumeAt = now + pause; + return pause; + } + + /** + * Records a successful send. + * @return true if this ended a back-off + */ + synchronized boolean succeeded() { + if (delaySeconds == 0) { + return false; + } + delaySeconds = 0; + resumeAt = Long.MIN_VALUE; + return true; + } +} diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java index ddb8809b..c4b84e9b 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java @@ -224,6 +224,16 @@ public abstract class ShareUsageBase */ protected long lostDataReset = LOST_DATA_RESET_DEFAULT; + /** + * Decides when sending may happen after failed sends. + */ + private final ShareUsageBackOff backOff; + + /** + * Released when the element starts closing, to end any pause at once. + */ + private final CountDownLatch closing = new CountDownLatch(1); + /** * Constructor * @param logger the logger to use @@ -392,6 +402,90 @@ protected ShareUsageBase( List> ignoreDataEvidenceFilter, String sessionCookieName, Tracker tracker) { + this( + logger, + sharePercentage, + minimumEntriesPerMessage, + maximumQueueSize, + addTimeout, + takeTimeout, + repeatEvidenceIntervalMinutes, + trackSession, + shareUsageUrl, + blockedHttpHeaders, + includedQueryStringParameters, + ignoreDataEvidenceFilter, + sessionCookieName, + tracker, + Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_DEFAULT, + Constants.SHARE_USAGE_FAILURES_WINDOW_SECONDS_DEFAULT, + Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_INITIAL_DELAY_SECONDS_DEFAULT, + Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MAX_DELAY_SECONDS_DEFAULT, + Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MULTIPLIER_DEFAULT); + } + + /** + * Constructor + * @param logger the logger to use + * @param sharePercentage the approximate proportion of requests to share. + * 1 = 100%, 0.5 = 50%, etc. + * @param minimumEntriesPerMessage the minimum number of request entries per + * message sent to 51Degrees + * @param maximumQueueSize the maximum number of items to hold in the queue + * at one time. This must be larger than minimum + * entries + * @param addTimeout the timeout in milliseconds to allow when attempting to + * add an item to the queue. If this timeout is exceeded + * then the item is not shared + * @param takeTimeout the timeout in milliseconds to allow when attempting + * to take an item to the queue + * @param repeatEvidenceIntervalMinutes the interval (in minutes) which is + * used to decide if repeat evidence is + * old enough to consider a new session + * @param trackSession set if the tracker should consider sessions in share + * usage + * @param shareUsageUrl the URL to send data to + * @param blockedHttpHeaders a list of the names of the HTTP headers that + * share usage should not send to 51Degrees + * @param includedQueryStringParameters a list of the names of query string + * parameters that share usage should + * send to 51Degrees + * @param ignoreDataEvidenceFilter the filter used to determine if an item + * of evidence should be ignored or not + * @param sessionCookieName the name of the cookie that contains the session + * id + * @param tracker the {@link Tracker} to use to determine if a given + * {@link FlowData} instance should be shared or not + * @param failuresToEnterRecovery the number of failed sends within the + * failures window that pauses sending + * @param failuresWindowSeconds the time in seconds over which failed sends + * are counted + * @param initialRecoveryDelaySeconds the length in seconds of the first + * pause + * @param maxRecoveryDelaySeconds the longest a pause can be, in seconds + * @param recoveryMultiplier what the pause is multiplied by each time the + * send tried when a pause ends fails + */ + protected ShareUsageBase( + Logger logger, + double sharePercentage, + int minimumEntriesPerMessage, + int maximumQueueSize, + int addTimeout, + int takeTimeout, + int repeatEvidenceIntervalMinutes, + boolean trackSession, + String shareUsageUrl, + List blockedHttpHeaders, + List includedQueryStringParameters, + List> ignoreDataEvidenceFilter, + String sessionCookieName, + Tracker tracker, + int failuresToEnterRecovery, + int failuresWindowSeconds, + double initialRecoveryDelaySeconds, + double maxRecoveryDelaySeconds, + double recoveryMultiplier) { super(logger, null); if (minimumEntriesPerMessage > maximumQueueSize) { @@ -406,6 +500,12 @@ protected ShareUsageBase( } evidenceCollection = new LinkedBlockingQueue<>(maximumQueueSize); + backOff = new ShareUsageBackOff( + failuresToEnterRecovery, + failuresWindowSeconds, + initialRecoveryDelaySeconds, + maxRecoveryDelaySeconds, + recoveryMultiplier); this.addTimeout = addTimeout; this.takeTimeout = takeTimeout; @@ -606,6 +706,9 @@ protected void managedResourcesCleanup() { // This is set before anything is submitted, so a consumer can never // observe it as false and then exit leaving a partial batch behind. stopping = true; + // Ends a pause at once, so closing never waits for one. Nothing is + // sent while paused. + closing.countDown(); // Queue a final consumer without consulting isRunning(). That check is // subject to a time-of-check/time-of-use race: a consumer that has // finished its loop with a sub-threshold remainder still reports as @@ -627,6 +730,13 @@ protected void managedResourcesCleanup() { // abandon its own wait immediately. logger.warn("Interrupted while awaiting close down"); } + if (evidenceCollection.isEmpty() == false && + backOff.remaining(System.currentTimeMillis()) > 0) { + logger.info( + "{} usage sharing entries were not sent, as sending was " + + "paused after failed sends when usage sharing closed.", + evidenceCollection.size()); + } } @Override @@ -784,6 +894,51 @@ private void submitSendData() { protected abstract void sendUsageData() throws HttpException; + /** + * Waits until sending may start again after failed sends. A consumer + * calls this before each message it sends. + * @return false if the element started closing while sending was paused, + * in which case nothing more should be sent + */ + protected boolean waitWhilePaused() { + long pause; + while ((pause = backOff.remaining(System.currentTimeMillis())) > 0) { + try { + if (closing.await(pause, TimeUnit.MILLISECONDS)) { + return false; + } + } catch (InterruptedException e) { + // Not re-interrupting, for the reason given in + // managedResourcesCleanup. The interrupt comes from closing. + return false; + } + } + return true; + } + + /** + * Records whether the remote service accepted a message, so that repeated + * failures pause sending and a success ends the pause. + * @param accepted true if the message was accepted + */ + protected void recordSendResult(boolean accepted) { + if (accepted) { + if (backOff.succeeded()) { + logger.info("Usage sharing resumed after a send succeeded."); + } + } else { + long pause = backOff.failed(System.currentTimeMillis()); + if (pause > 0) { + logger.warn( + "Usage sharing paused for {} seconds after failed " + + "sends. Usage data is still queued, up to the " + + "maximum queue size, and sending starts again when " + + "the pause ends.", + pause / 1000.0); + } + } + } + /** * Get engine Version No. * @return engine version no diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilder.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilder.java index 56a96e9a..21e06859 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilder.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilder.java @@ -129,7 +129,12 @@ public ShareUsageElement build() throws IOException { includedQueryStringParameters, ignoreDataEvidenceFilter, sessionCookieName, - null); + null, + failuresToEnterRecovery, + failuresWindowSeconds, + initialRecoveryDelaySeconds, + maxRecoveryDelaySeconds, + recoveryMultiplier); if (Objects.nonNull(this.httpClient)) { sue.httpClient = httpClient; } else if (Objects.nonNull(dataUploader)){ diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilderBase.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilderBase.java index e376660c..3ad5d30f 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilderBase.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilderBase.java @@ -59,6 +59,11 @@ public abstract class ShareUsageBuilderBase { protected final List includedQueryStringParameters = new ArrayList<>(); protected final List> ignoreDataEvidenceFilter = new ArrayList<>(); protected boolean trackSession; + protected int failuresToEnterRecovery = Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_DEFAULT; + protected int failuresWindowSeconds = Constants.SHARE_USAGE_FAILURES_WINDOW_SECONDS_DEFAULT; + protected double initialRecoveryDelaySeconds = Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_INITIAL_DELAY_SECONDS_DEFAULT; + protected double maxRecoveryDelaySeconds = Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MAX_DELAY_SECONDS_DEFAULT; + protected double recoveryMultiplier = Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MULTIPLIER_DEFAULT; /** * Constructor @@ -420,6 +425,92 @@ public ShareUsageBuilderBase setTrackSession(boolean track) { return this; } + /** + * Set the number of failed sends within the failures window that pauses + * sending. Usage data is still queued while sending is paused, up to the + * maximum queue size. + * @param failuresToEnterRecovery the number of failed sends, from 1 to 100 + * @return this builder + */ + @DefaultValue(intValue = Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_DEFAULT) + public ShareUsageBuilderBase setFailuresToEnterRecovery( + int failuresToEnterRecovery) { + if (failuresToEnterRecovery < Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MIN || + failuresToEnterRecovery > Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MAX) { + throw new IllegalArgumentException( + "failuresToEnterRecovery must be from " + + Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MIN + " to " + + Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MAX + + ". Received: " + failuresToEnterRecovery + "."); + } + this.failuresToEnterRecovery = failuresToEnterRecovery; + return this; + } + + /** + * Set the time in seconds over which failed sends are counted. + * @param failuresWindowSeconds the time in seconds, 1 or more + * @return this builder + */ + @DefaultValue(intValue = Constants.SHARE_USAGE_FAILURES_WINDOW_SECONDS_DEFAULT) + public ShareUsageBuilderBase setFailuresWindowSeconds( + int failuresWindowSeconds) { + if (failuresWindowSeconds < 1) { + throw new IllegalArgumentException( + "failuresWindowSeconds must be 1 or more."); + } + this.failuresWindowSeconds = failuresWindowSeconds; + return this; + } + + /** + * Set the length in seconds of the first pause after failed sends. + * @param initialDelaySeconds the length in seconds, more than 0 + * @return this builder + */ + @DefaultValue(doubleValue = Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_INITIAL_DELAY_SECONDS_DEFAULT) + public ShareUsageBuilderBase setInitialRecoveryDelaySeconds( + double initialDelaySeconds) { + if ((initialDelaySeconds > 0) == false) { + throw new IllegalArgumentException( + "initialDelaySeconds must be more than 0."); + } + this.initialRecoveryDelaySeconds = initialDelaySeconds; + return this; + } + + /** + * Set the longest a pause can be, in seconds. + * @param maxDelaySeconds the length in seconds, more than 0 + * @return this builder + */ + @DefaultValue(doubleValue = Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MAX_DELAY_SECONDS_DEFAULT) + public ShareUsageBuilderBase setMaxRecoveryDelaySeconds( + double maxDelaySeconds) { + if ((maxDelaySeconds > 0) == false) { + throw new IllegalArgumentException( + "maxDelaySeconds must be more than 0."); + } + this.maxRecoveryDelaySeconds = maxDelaySeconds; + return this; + } + + /** + * Set what the pause is multiplied by each time the send tried when a + * pause ends fails. 2 doubles the pause each time. + * @param multiplier the multiplier, 1 or more + * @return this builder + */ + @DefaultValue(doubleValue = Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MULTIPLIER_DEFAULT) + public ShareUsageBuilderBase setRecoveryMultiplier(double multiplier) { + if ((multiplier >= 1) == false) { + throw new IllegalArgumentException( + "multiplier must be 1 or more."); + } + this.recoveryMultiplier = multiplier; + return this; + } + /** * Create the {@link ShareUsageElement}. * @return the newly created target instance diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java index c56e032f..79d85474 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java @@ -46,6 +46,11 @@ import java.util.zip.GZIPOutputStream; import static fiftyone.pipeline.engines.fiftyone.flowelements.Constants.SHARE_USAGE_DEFAULT_HTTP_POST_TIMEOUT; +import static fiftyone.pipeline.engines.fiftyone.flowelements.Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_INITIAL_DELAY_SECONDS_DEFAULT; +import static fiftyone.pipeline.engines.fiftyone.flowelements.Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MAX_DELAY_SECONDS_DEFAULT; +import static fiftyone.pipeline.engines.fiftyone.flowelements.Constants.SHARE_USAGE_EXPONENTIAL_BACKOFF_MULTIPLIER_DEFAULT; +import static fiftyone.pipeline.engines.fiftyone.flowelements.Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_DEFAULT; +import static fiftyone.pipeline.engines.fiftyone.flowelements.Constants.SHARE_USAGE_FAILURES_WINDOW_SECONDS_DEFAULT; import static fiftyone.pipeline.engines.fiftyone.flowelements.Constants.SHARE_USAGE_MAX_EVIDENCE_LENGTH; /** @@ -309,6 +314,90 @@ public class ShareUsageElement extends ShareUsageBase { List> ignoreDataEvidenceFilter, String sessionCookieName, Tracker tracker) { + this( + logger, + sharePercentage, + minimumEntriesPerMessage, + maximumQueueSize, + addTimeout, + takeTimeout, + repeatEvidenceIntervalMinutes, + trackSession, + shareUsageUrl, + blockedHttpHeaders, + includedQueryStringParameters, + ignoreDataEvidenceFilter, + sessionCookieName, + tracker, + SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_DEFAULT, + SHARE_USAGE_FAILURES_WINDOW_SECONDS_DEFAULT, + SHARE_USAGE_EXPONENTIAL_BACKOFF_INITIAL_DELAY_SECONDS_DEFAULT, + SHARE_USAGE_EXPONENTIAL_BACKOFF_MAX_DELAY_SECONDS_DEFAULT, + SHARE_USAGE_EXPONENTIAL_BACKOFF_MULTIPLIER_DEFAULT); + } + + /** + * Constructor + * + * @param logger the logger to use + * @param sharePercentage the approximate proportion of requests to share. + * 1 = 100%, 0.5 = 50%, etc. + * @param minimumEntriesPerMessage the minimum number of request entries per + * message sent to 51Degrees + * @param maximumQueueSize the maximum number of items to hold in the queue + * at one time. This must be larger than minimum + * entries + * @param addTimeout the timeout in milliseconds to allow when attempting to + * add an item to the queue. If this timeout is exceeded + * then the item is not shared + * @param takeTimeout the timeout in milliseconds to allow when attempting + * to take an item to the queue + * @param repeatEvidenceIntervalMinutes the interval (in minutes) which is + * used to decide if repeat evidence is + * old enough to consider a new session + * @param trackSession set if the tracker should consider sessions in share + * usage + * @param shareUsageUrl the URL to send data to + * @param blockedHttpHeaders a list of the names of the HTTP headers that + * share usage should not send to 51Degrees + * @param includedQueryStringParameters a list of the names of query string + * parameters that share usage should + * send to 51Degrees + * @param ignoreDataEvidenceFilter the filter used to determine if an item + * of evidence should be ignored or not + * @param sessionCookieName the name of the cookie that contains the session + * id + * @param tracker the {@link Tracker} to use to determine if a given + * {@link FlowData} instance should be shared or not + * @param failuresToEnterRecovery the number of failed sends within the failures + * window that pauses sending + * @param failuresWindowSeconds the time in seconds over which failed sends are + * counted + * @param initialRecoveryDelaySeconds the length in seconds of the first pause + * @param maxRecoveryDelaySeconds the longest a pause can be, in seconds + * @param recoveryMultiplier what the pause is multiplied by each time the send + * tried when a pause ends fails + */ + ShareUsageElement( + Logger logger, + double sharePercentage, + int minimumEntriesPerMessage, + int maximumQueueSize, + int addTimeout, + int takeTimeout, + int repeatEvidenceIntervalMinutes, + boolean trackSession, + String shareUsageUrl, + List blockedHttpHeaders, + List includedQueryStringParameters, + List> ignoreDataEvidenceFilter, + String sessionCookieName, + Tracker tracker, + int failuresToEnterRecovery, + int failuresWindowSeconds, + double initialRecoveryDelaySeconds, + double maxRecoveryDelaySeconds, + double recoveryMultiplier) { super( logger, sharePercentage, @@ -323,7 +412,12 @@ public class ShareUsageElement extends ShareUsageBase { includedQueryStringParameters, ignoreDataEvidenceFilter, sessionCookieName, - tracker); + tracker, + failuresToEnterRecovery, + failuresWindowSeconds, + initialRecoveryDelaySeconds, + maxRecoveryDelaySeconds, + recoveryMultiplier); headers.put("Content-Type", "text/xml; charset=utf-8"); // assume default GZIP connection @@ -336,7 +430,8 @@ public class ShareUsageElement extends ShareUsageBase { } /** - * Take data from the queue until there are fewer entries than the minimum batch size + * Take data from the queue until there are fewer entries than the minimum + * batch size, waiting out any pause after failed sends before each message */ @Override protected void sendUsageData() { @@ -345,6 +440,10 @@ protected void sendUsageData() { List allData = new ArrayList<>(); // drain queue while minimum entries in it do { + // Closing while paused ends the wait, and nothing more is sent. + if (waitWhilePaused() == false) { + break; + } logger.debug(threadMarker, "Queue size is {}", evidenceCollection.size()); try { allData.clear(); @@ -362,7 +461,13 @@ protected void sendUsageData() { // drained everything - and there is no point posting a document // with no entries in it. if (allData.isEmpty() == false) { - sendAsXML(allData); + boolean accepted = false; + try { + sendAsXML(allData); + accepted = true; + } finally { + recordSendResult(accepted); + } } } catch (InterruptedException interruptedException) { // Deliberately not re-interrupting: the interrupt comes from diff --git a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOffTests.java b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOffTests.java new file mode 100644 index 00000000..6cb71eb7 --- /dev/null +++ b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOffTests.java @@ -0,0 +1,184 @@ +/* ********************************************************************* + * This Original Work is copyright of 51 Degrees Mobile Experts Limited. + * Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, + * Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. + * + * This Original Work is licensed under the European Union Public Licence + * (EUPL) v.1.2 and is subject to its terms as set out below. + * + * If a copy of the EUPL was not distributed with this file, You can obtain + * one at https://opensource.org/licenses/EUPL-1.2. + * + * The 'Compatible Licences' set out in the Appendix to the EUPL (as may be + * amended by the European Commission) shall be deemed incompatible for + * the purposes of the Work and the provisions of the compatibility + * clause in Article 5 of the EUPL shall not apply. + * + * If using the Work as, or as part of, a network application, by + * including the attribution notice(s) required under Article 5 of the EUPL + * in the end user terms of the application under an appropriate heading, + * such notice(s) shall fulfill the requirements of that article. + * ********************************************************************* */ + +package fiftyone.pipeline.engines.fiftyone.flowelements; + +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * The rules that decide when usage sharing may send after failed sends, + * with each test setting the time, and the builder settings for them. + */ +public class ShareUsageBackOffTests { + + private static final long START = 1767225600000L; + + private static long at(double seconds) { + return START + Math.round(seconds * 1000); + } + + /** + * Three failures within a ten second window start a two second pause, + * rising by two each time up to five. + */ + private static ShareUsageBackOff create( + int failuresToEnterRecovery, + double initialDelaySeconds, + double maxDelaySeconds) { + return new ShareUsageBackOff( + failuresToEnterRecovery, + 10, + initialDelaySeconds, + maxDelaySeconds, + 2); + } + + private static ShareUsageBackOff create() { + return create(3, 2, 5); + } + + @Test + public void ShareUsageBackOff_PausesAfterFailuresWithinWindow() { + ShareUsageBackOff backOff = create(); + + assertEquals(0, backOff.failed(at(0))); + assertEquals(0, backOff.failed(at(1))); + assertEquals(0, backOff.remaining(at(1)), + "Sending paused before enough sends failed."); + assertFalse(backOff.isBackingOff()); + + assertEquals(2000, backOff.failed(at(2))); + assertTrue(backOff.isBackingOff()); + assertEquals(1000, backOff.remaining(at(3))); + assertEquals(0, backOff.remaining(at(4))); + } + + @Test + public void ShareUsageBackOff_FailuresOutsideWindowDoNotCount() { + ShareUsageBackOff backOff = create(); + + backOff.failed(at(0)); + backOff.failed(at(1)); + + assertEquals(0, backOff.failed(at(11.5)), + "Failures older than the window were counted."); + assertFalse(backOff.isBackingOff()); + } + + /** + * The window counts failures, not failures in a row, so a high failure + * rate pauses sending even when some sends succeed. + */ + @Test + public void ShareUsageBackOff_SuccessesBetweenFailuresDoNotReset() { + ShareUsageBackOff backOff = create(); + + backOff.failed(at(0)); + assertFalse(backOff.succeeded()); + backOff.failed(at(1)); + assertFalse(backOff.succeeded()); + + assertEquals(2000, backOff.failed(at(2))); + } + + @Test + public void ShareUsageBackOff_FailedSendAfterPauseMultipliesDelay() { + ShareUsageBackOff backOff = create(1, 2, 5); + + assertEquals(2000, backOff.failed(at(0))); + assertEquals(4000, backOff.failed(at(2))); + assertEquals(5000, backOff.failed(at(6)), + "The delay grew past the maximum."); + assertEquals(5000, backOff.failed(at(11))); + assertEquals(5000, backOff.remaining(at(11))); + } + + @Test + public void ShareUsageBackOff_SuccessEndsBackOff() { + ShareUsageBackOff backOff = create(2, 2, 5); + backOff.failed(at(0)); + backOff.failed(at(1)); + backOff.failed(at(3)); + + assertTrue(backOff.succeeded(), + "A success did not end the back-off."); + assertFalse(backOff.isBackingOff()); + assertEquals(0, backOff.remaining(at(4))); + + // The next pause needs the full number of failures again, and starts + // from the initial delay. + assertEquals(0, backOff.failed(at(8))); + assertEquals(2000, backOff.failed(at(9))); + } + + @Test + public void ShareUsageBackOff_MaximumBelowInitialLimitsFirstPause() { + ShareUsageBackOff backOff = create(1, 10, 3); + + assertEquals(3000, backOff.failed(at(0))); + } + + @Test + public void ShareUsageBackOff_InvalidSettingsAreRejected() { + assertThrows(IllegalArgumentException.class, + () -> new ShareUsageBackOff(0, 10, 2, 5, 2)); + assertThrows(IllegalArgumentException.class, + () -> new ShareUsageBackOff(1, 0, 2, 5, 2)); + assertThrows(IllegalArgumentException.class, + () -> new ShareUsageBackOff(1, 10, 0, 5, 2)); + assertThrows(IllegalArgumentException.class, + () -> new ShareUsageBackOff(1, 10, Double.NaN, 5, 2)); + assertThrows(IllegalArgumentException.class, + () -> new ShareUsageBackOff(1, 10, 2, 0, 2)); + assertThrows(IllegalArgumentException.class, + () -> new ShareUsageBackOff(1, 10, 2, 5, 0.5)); + } + + /** + * The builder settings that pause sending after failed sends reject + * values the pause cannot work with. + */ + @Test + public void ShareUsageBuilder_RecoverySettings_RejectValuesOutOfRange() { + ShareUsageBuilder builder = + new ShareUsageBuilder(LoggerFactory.getILoggerFactory()); + + assertThrows(IllegalArgumentException.class, + () -> builder.setFailuresToEnterRecovery(0)); + assertThrows(IllegalArgumentException.class, + () -> builder.setFailuresToEnterRecovery(101)); + assertThrows(IllegalArgumentException.class, + () -> builder.setFailuresWindowSeconds(0)); + assertThrows(IllegalArgumentException.class, + () -> builder.setInitialRecoveryDelaySeconds(0)); + assertThrows(IllegalArgumentException.class, + () -> builder.setMaxRecoveryDelaySeconds(-1)); + assertThrows(IllegalArgumentException.class, + () -> builder.setRecoveryMultiplier(0.5)); + } +} diff --git a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java index da10ee1e..a6a0b2e1 100644 --- a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java +++ b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java @@ -161,6 +161,40 @@ public void init() throws Exception { element.dataUploader = uploader; } + /** + * Replaces the element with one that has its own back-off settings. The + * failures window is long enough that every failure in a test counts. + */ + private void useBackOff( + int failuresToEnterRecovery, + double initialDelaySeconds, + double maxDelaySeconds, + int maximumQueueSize) throws Exception { + element.close(); + element = new ShareUsageElement( + LoggerFactory.getLogger(ShareUsageElement.class), + 1, + 1, + maximumQueueSize, + 100, + 100, + 1, + true, + "http://localhost/usage", + new ArrayList<>(), + new ArrayList<>(), + new ArrayList<>(), + DEFAULT_SESSION_COOKIE_NAME, + new ShareAll(), + failuresToEnterRecovery, + 100, + initialDelaySeconds, + maxDelaySeconds, + 2); + element.addPipeline(pipeline); + element.dataUploader = uploader; + } + @AfterEach public void cleanUp() throws Exception { element.close(); @@ -262,4 +296,63 @@ public void ShareUsageElement_BackOff_CloseWhilePausedDoesNotWaitOrSend() assertEquals(DEFAULT_FAILURES, uploader.count(), "A send was made during the pause or when closing."); } + + /** + * Each failed send after a pause multiplies the delay, and the delay + * stops growing at the maximum. + */ + @Test + public void ShareUsageElement_BackOff_DelayGrowsToMaximum() + throws Exception { + useBackOff(1, 0.5, 1.0, 100); + for (int i = 0; i < 40; i++) { + process(); + } + + waitFor( + () -> uploader.count() >= 4, + () -> "Fewer than four sends were made."); + List times = uploader.times(); + long first = times.get(1) - times.get(0); + long second = times.get(2) - times.get(1); + long third = times.get(3) - times.get(2); + String shown = first + ", " + second + ", " + third + " ms"; + assertTrue(first >= 400, + "The first pause was not the initial delay. Gaps: " + shown); + assertTrue(second >= 900, + "The second pause was not doubled. Gaps: " + shown); + assertTrue(third >= 900 && third < 1900, + "The third pause was not held at the maximum. Gaps: " + shown); + } + + /** + * When the queue fills during a pause, new evidence is dropped and what + * was queued first is sent once sending resumes. + */ + @Test + public void ShareUsageElement_BackOff_FullQueueDropsNewEvidence() + throws Exception { + useBackOff(1, 2, 2, 4); + failSends(1); + + List kept = new ArrayList<>(); + for (int i = 0; i < 4; i++) { + kept.add(process()); + } + List dropped = new ArrayList<>(); + for (int i = 0; i < 6; i++) { + dropped.add(process()); + } + uploader.code = 200; + + waitFor( + () -> uploader.accepted().containsAll(kept), + () -> "The evidence queued during the pause was not all sent. " + + "Sent: " + uploader.accepted()); + Thread.sleep(500); + for (String ip : dropped) { + assertTrue(uploader.accepted().contains(ip) == false, + "Evidence processed when the queue was full was sent."); + } + } } From 4c53537dbef038e2016ef7c60f1525ec05c884be Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Sat, 7 Nov 2026 07:12:05 +0000 Subject: [PATCH 03/16] Check the whole range of failures to enter recovery in the back-off The builder setter rejects a value outside 1 to 100, and the back-off now does too, so a class derived from ShareUsageBase gets the same check through its constructor. --- .../fiftyone/flowelements/ShareUsageBackOff.java | 10 ++++++++-- .../fiftyone/flowelements/ShareUsageBackOffTests.java | 2 ++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java index 3cf13cde..8cc06753 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java @@ -63,9 +63,15 @@ class ShareUsageBackOff { double initialDelaySeconds, double maxDelaySeconds, double multiplier) { - if (failuresToEnterRecovery < 1) { + if (failuresToEnterRecovery < + Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MIN || + failuresToEnterRecovery > + Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MAX) { throw new IllegalArgumentException( - "failuresToEnterRecovery must be 1 or more."); + "failuresToEnterRecovery must be from " + + Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MIN + + " to " + + Constants.SHARE_USAGE_FAILURES_TO_ENTER_RECOVERY_MAX + "."); } if (failuresWindowSeconds < 1) { throw new IllegalArgumentException( diff --git a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOffTests.java b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOffTests.java index 6cb71eb7..b9716f9f 100644 --- a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOffTests.java +++ b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOffTests.java @@ -147,6 +147,8 @@ public void ShareUsageBackOff_MaximumBelowInitialLimitsFirstPause() { public void ShareUsageBackOff_InvalidSettingsAreRejected() { assertThrows(IllegalArgumentException.class, () -> new ShareUsageBackOff(0, 10, 2, 5, 2)); + assertThrows(IllegalArgumentException.class, + () -> new ShareUsageBackOff(101, 10, 2, 5, 2)); assertThrows(IllegalArgumentException.class, () -> new ShareUsageBackOff(1, 0, 2, 5, 2)); assertThrows(IllegalArgumentException.class, From 033155f6ca5ed16a3d23b0ed286225acc5f86f0b Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Sat, 7 Nov 2026 07:14:53 +0000 Subject: [PATCH 04/16] Keep history out of the comments in the changed send code The comments in the functions this change touches described what the code used to do. They now say only what the code does and why. --- .../fiftyone/flowelements/ShareUsageBase.java | 14 ++++++-------- .../fiftyone/flowelements/ShareUsageElement.java | 2 +- 2 files changed, 7 insertions(+), 9 deletions(-) diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java index c4b84e9b..f01450cb 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java @@ -709,14 +709,12 @@ protected void managedResourcesCleanup() { // Ends a pause at once, so closing never waits for one. Nothing is // sent while paused. closing.countDown(); - // Queue a final consumer without consulting isRunning(). That check is - // subject to a time-of-check/time-of-use race: a consumer that has - // finished its loop with a sub-threshold remainder still reports as - // running until its Future flips to done, which used to stop close from - // queueing the drain that would have sent the remainder. The executor is - // single threaded, so the task queued here cannot run alongside a - // consumer that is already in flight - it runs after it - and shutdown() - // still allows an already submitted task to run to completion. + // Queue a final consumer without consulting isRunning(). A consumer + // that has finished its loop with a sub-threshold remainder still + // reports as running until its Future is done, so checking would skip + // the drain that sends the remainder. The executor is single threaded, + // so the task queued here runs after any consumer already in flight, + // and shutdown() still allows an already submitted task to finish. submitSendData(); try { executor.shutdown(); diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java index 79d85474..730a5970 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java @@ -447,7 +447,7 @@ protected void sendUsageData() { logger.debug(threadMarker, "Queue size is {}", evidenceCollection.size()); try { allData.clear(); - // for historical reasons max entries set to 2* + // A message holds at most twice the minimum entries. while (allData.size() < minEntriesPerMessage * 2) { ShareUsageData currentData = evidenceCollection.poll(takeTimeout, TimeUnit.MILLISECONDS); // no data available From a510270651648713bd93d936b6193dac3ff6629f Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 19:50:02 +0000 Subject: [PATCH 05/16] Test the usage sharing options, the case of keys and the tracker's cache --- .../flowelements/ShareUsageOptionsTests.java | 273 ++++++++++++++++++ 1 file changed, 273 insertions(+) create mode 100644 pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageOptionsTests.java diff --git a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageOptionsTests.java b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageOptionsTests.java new file mode 100644 index 00000000..ef2e2610 --- /dev/null +++ b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageOptionsTests.java @@ -0,0 +1,273 @@ +/* ********************************************************************* + * This Original Work is copyright of 51 Degrees Mobile Experts Limited. + * Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, + * Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. + * + * This Original Work is licensed under the European Union Public Licence + * (EUPL) v.1.2 and is subject to its terms as set out below. + * + * If a copy of the EUPL was not distributed with this file, You can obtain + * one at https://opensource.org/licenses/EUPL-1.2. + * + * The 'Compatible Licences' set out in the Appendix to the EUPL (as may be + * amended by the European Commission) shall be deemed incompatible for + * the purposes of the Work and the provisions of the compatibility + * clause in Article 5 of the EUPL shall not apply. + * + * If using the Work as, or as part of, a network application, by + * including the attribution notice(s) required under Article 5 of the EUPL + * in the end user terms of the application under an appropriate heading, + * such notice(s) shall fulfill the requirements of that article. + * ********************************************************************* */ + +package fiftyone.pipeline.engines.fiftyone.flowelements; + +import fiftyone.caching.Cache; +import fiftyone.caching.CacheBuilder; +import fiftyone.caching.LruPutCache; +import fiftyone.pipeline.core.configuration.ElementOptions; +import fiftyone.pipeline.core.configuration.PipelineOptions; +import fiftyone.pipeline.core.data.FlowData; +import fiftyone.pipeline.core.flowelements.Pipeline; +import fiftyone.pipeline.core.flowelements.PipelineBuilder; +import fiftyone.pipeline.engines.configuration.CacheConfiguration; +import fiftyone.pipeline.engines.fiftyone.data.EvidenceKeyFilterShareUsage; +import fiftyone.pipeline.engines.fiftyone.trackers.ShareUsageTracker; +import fiftyone.pipeline.engines.services.DataUploader; +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; +import org.w3c.dom.Document; +import org.xml.sax.InputSource; + +import javax.xml.parsers.DocumentBuilderFactory; +import javax.xml.xpath.XPathFactory; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.OutputStream; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import static fiftyone.pipeline.engines.Constants.DEFAULT_SESSION_COOKIE_NAME; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests the usage sharing options "Share all Evidence" and "Share all query + * string parameters" from a configuration file, the case of the keys the + * element reads and the number of caches a tracker holds. + * @see Specification + */ +public class ShareUsageOptionsTests { + + /** + * Keeps what the element sends so a test can read it. + */ + private static class CapturingUploader implements DataUploader { + private final ByteArrayOutputStream sent = new ByteArrayOutputStream(); + + @Override + public OutputStream getOutputStream() { + return sent; + } + + @Override + public int getResponseCode() { + return 200; + } + + Document document() throws Exception { + return DocumentBuilderFactory.newInstance() + .newDocumentBuilder() + .parse(new InputSource( + new ByteArrayInputStream(sent.toByteArray()))); + } + } + + /** + * Counts the caches the tracker asks for. + */ + private static class CountingCacheBuilder implements CacheBuilder { + final AtomicInteger built = new AtomicInteger(); + private final LruPutCache.Builder inner = new LruPutCache.Builder(); + + @Override + public Cache build(Cache c, int cacheSize) { + built.incrementAndGet(); + return inner.build(c, cacheSize); + } + } + + /** + * Options for a usage sharing element that sends every request at once, + * as a configuration file holds them. + */ + private static PipelineOptions options(String... nameAndValue) { + ElementOptions element = new ElementOptions(); + element.builderName = "ShareUsage"; + element.buildParameters.put("SharePercentage", "1"); + element.buildParameters.put("MinimumEntriesPerMessage", "1"); + element.buildParameters.put("RepeatEvidenceIntervalMinutes", "0"); + element.buildParameters.put("BlockedHttpHeader", "x-blocked"); + for (int i = 0; i < nameAndValue.length; i += 2) { + element.buildParameters.put(nameAndValue[i], nameAndValue[i + 1]); + } + PipelineOptions options = new PipelineOptions(); + options.elements.add(element); + return options; + } + + private static Map evidence() { + Map evidence = new HashMap<>(); + evidence.put("header.user-agent", "agent"); + evidence.put("header.x-blocked", "blocked"); + evidence.put("header.cookie", "a=b"); + evidence.put("cookie.plain", "cookie"); + evidence.put("session.plain", "session"); + evidence.put("query.plain", "query"); + evidence.put("query.id.email", "a@example.com"); + evidence.put("id.email", "a@example.com"); + evidence.put("id.salt", "salt"); + evidence.put("server.request-path", "/path"); + evidence.put("server.request-query", "?plain=query"); + return evidence; + } + + /** + * Builds the element from the options, gives it the Evidence and returns + * what it sends. + */ + private static Document send( + PipelineOptions options, + Map evidence) throws Exception { + CapturingUploader uploader = new CapturingUploader(); + try (Pipeline pipeline = new PipelineBuilder( + LoggerFactory.getILoggerFactory()) + .buildFromConfiguration(options)) { + ShareUsageElement element = + pipeline.getElement(ShareUsageElement.class); + assertNotNull(element); + element.dataUploader = uploader; + try (FlowData data = pipeline.createFlowData()) { + data.addEvidence(evidence); + element.process(data); + assertNotNull(element.getSendDataFuture()); + element.getSendDataFuture().get(); + } + } + return uploader.document(); + } + + private static boolean shared( + Document sent, + String category, + String name) throws Exception { + return !XPathFactory.newInstance().newXPath().evaluate( + "//Device/" + category + "[@Name='" + name + "']", + sent).isEmpty(); + } + + @Test + public void defaults_ShareOnlyWhatTheRulesAllow() throws Exception { + Document sent = send(options(), evidence()); + + assertTrue(shared(sent, "header", "user-agent")); + assertFalse(shared(sent, "header", "x-blocked")); + assertFalse(shared(sent, "cookie", "plain")); + assertFalse(shared(sent, "session", "plain")); + assertFalse(shared(sent, "query", "plain")); + } + + @Test + public void shareAllEvidence_FromConfiguration_SharesEverythingButNeverShared() + throws Exception { + Document sent = send(options("ShareAllEvidence", "true"), evidence()); + + assertTrue(shared(sent, "header", "user-agent")); + assertTrue(shared(sent, "header", "x-blocked")); + assertTrue(shared(sent, "cookie", "plain")); + assertTrue(shared(sent, "session", "plain")); + assertTrue(shared(sent, "query", "plain")); + assertFalse(shared(sent, "query", "id.email")); + assertFalse(shared(sent, "id", "email")); + assertFalse(shared(sent, "id", "salt")); + assertFalse(shared(sent, "header", "cookie")); + assertFalse(shared(sent, "server", "request-path")); + assertFalse(shared(sent, "server", "request-query")); + } + + @Test + public void shareAllEvidence_False_FromConfiguration_KeepsTheRules() + throws Exception { + Document sent = send(options("ShareAllEvidence", "false"), evidence()); + + assertFalse(shared(sent, "header", "x-blocked")); + assertFalse(shared(sent, "cookie", "plain")); + assertFalse(shared(sent, "query", "plain")); + } + + @Test + public void shareAllQueryStringParameters_FromConfiguration_SharesEveryParameter() + throws Exception { + Document sent = send( + options("ShareAllQueryStringParameters", "true"), + evidence()); + + assertTrue(shared(sent, "query", "plain")); + assertFalse(shared(sent, "query", "id.email")); + assertFalse(shared(sent, "server", "request-query")); + // The other rules still apply. + assertFalse(shared(sent, "header", "x-blocked")); + assertFalse(shared(sent, "cookie", "plain")); + assertFalse(shared(sent, "session", "plain")); + assertTrue(shared(sent, "header", "user-agent")); + } + + @Test + public void shareAllQueryStringParameters_False_FromConfiguration_KeepsTheRules() + throws Exception { + Document sent = send( + options("ShareAllQueryStringParameters", "false"), + evidence()); + + assertFalse(shared(sent, "query", "plain")); + } + + @Test + public void keysAreReadWithoutRegardToCase() throws Exception { + Map evidence = new HashMap<>(); + evidence.put("Server.Client-IP", "1.2.3.4"); + evidence.put("Query.Session-ID", "session-1"); + evidence.put("Query.Sequence", "5"); + + Document sent = send(options(), evidence); + + assertEquals("1.2.3.4", + sent.getElementsByTagName("ClientIP").item(0).getTextContent()); + assertEquals("session-1", + sent.getElementsByTagName("SessionId").item(0).getTextContent()); + assertEquals("5", + sent.getElementsByTagName("Sequence").item(0).getTextContent()); + } + + @Test + public void tracker_BuildsOneCache() { + CountingCacheBuilder builder = new CountingCacheBuilder(); + + try (ShareUsageTracker tracker = new ShareUsageTracker( + new CacheConfiguration(builder, 10), + 1000, + new EvidenceKeyFilterShareUsage( + new ArrayList(), + new ArrayList(), + false, + DEFAULT_SESSION_COOKIE_NAME))) { + assertEquals(1, builder.built.get()); + } catch (java.io.IOException e) { + throw new IllegalStateException(e); + } + } +} From f29c1a9548cea1e2d0996c6cd7999b110f7574c3 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 19:53:54 +0000 Subject: [PATCH 06/16] Add the usage sharing options to share all Evidence and all query string parameters, read keys without regard to case, and keep one tracker cache --- .../data/EvidenceKeyFilterShareUsage.java | 59 +++++++++++- .../EvidenceKeyFilterShareUsageTracker.java | 32 +++++++ .../fiftyone/flowelements/ShareUsageBase.java | 91 ++++++++++++++++++- .../flowelements/ShareUsageBuilder.java | 4 +- .../flowelements/ShareUsageBuilderBase.java | 52 +++++++++++ .../flowelements/ShareUsageElement.java | 77 +++++++++++++++- .../fiftyone/trackers/ShareUsageTracker.java | 42 ++------- .../flowelements/ShareUsageOptionsTests.java | 9 ++ .../engines/caching/DataKeyedCacheBase.java | 16 +++- 9 files changed, 336 insertions(+), 46 deletions(-) diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsage.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsage.java index 07a21856..d3c09061 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsage.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsage.java @@ -47,9 +47,12 @@ * {@link Constants#FIFTYONE_COOKIE_PREFIX} or has been configured to be * shared using the constructor parameter. * 5) Not a header, cookie, session value or query string parameter. + * When all Evidence is shared, everything is included. When all query string + * parameters are shared, every query string parameter is included. * The prefix of a key is everything before its first '.', and the field * everything after it. An email address, the salt for hashed email - * identifiers, the query string and the path are never included. + * identifiers, the query string, the path and the Cookie header are never + * included, whatever the configuration. */ public class EvidenceKeyFilterShareUsage implements EvidenceKeyFilter { @@ -65,6 +68,17 @@ public class EvidenceKeyFilterShareUsage implements EvidenceKeyFilter { */ private final String sessionCookieName; + /** + * If true then every key is included, apart from those that are never + * shared. + */ + private final boolean shareAllEvidence; + + /** + * If true then every query string parameter is included. + */ + private final boolean shareAllQueryStringParameters; + /** * The content of HTTP headers in this array will not be included in the * request information sent to 51degrees. Any header names added here are @@ -113,6 +127,40 @@ public EvidenceKeyFilterShareUsage( List includedQueryStringParams, boolean includeSession, String sessionCookieName) { + this( + blockedHttpHeaders, + includedQueryStringParams, + includeSession, + sessionCookieName, + false, + false); + } + + /** + * Constructor. + * @param blockedHttpHeaders a list of the names of the HTTP headers that + * share usage should not send to 51Degrees + * @param includedQueryStringParams a list of the names of query string + * parameters that share usage should send + * to 51Degrees + * @param includeSession if true then the session cookie will be included in + * the filter + * @param sessionCookieName the name of the cookie that contains the session + * id + * @param shareAllEvidence if true then all Evidence is included, apart from + * the keys that are never shared + * @param shareAllQueryStringParameters if true then all query string + * parameters are included + */ + public EvidenceKeyFilterShareUsage( + List blockedHttpHeaders, + List includedQueryStringParams, + boolean includeSession, + String sessionCookieName, + boolean shareAllEvidence, + boolean shareAllQueryStringParameters) { + this.shareAllEvidence = shareAllEvidence; + this.shareAllQueryStringParameters = shareAllQueryStringParameters; // The Cookie header carries every cookie, including the ones the // cookie rule leaves out. this.blockedHttpHeaders.add("cookies"); @@ -144,6 +192,10 @@ public boolean include(String key) { } } + if (shareAllEvidence) { + return true; + } + boolean result; int separator = lowerKey.indexOf(EVIDENCE_SEPERATOR); if (separator > 0) { @@ -169,8 +221,9 @@ public boolean include(String key) { case EVIDENCE_QUERY_PREFIX: // Only include query string parameters that start with the // 51Degrees prefix or have been specified in the - // constructor. - result = field.startsWith(Constants.FIFTYONE_COOKIE_PREFIX) || + // constructor, unless all of them are shared. + result = shareAllQueryStringParameters || + field.startsWith(Constants.FIFTYONE_COOKIE_PREFIX) || includedQueryStringParams.contains(field); break; default: diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsageTracker.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsageTracker.java index 9240b326..61c2c348 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsageTracker.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/data/EvidenceKeyFilterShareUsageTracker.java @@ -57,6 +57,38 @@ public EvidenceKeyFilterShareUsageTracker( sessionCookieName); } + /** + * Constructor. + * @param blockedHttpHeaders a list of the names of the HTTP headers that + * share usage should not send to 51Degrees + * @param includedQueryStringParams a list of the names of query string + * parameters that share usage should send + * to 51Degrees + * @param includeSession if true then the session cookie will be included in + * the filter + * @param sessionCookieName the name of the cookie that contains the session + * id + * @param shareAllEvidence if true then all Evidence is included, apart from + * the keys that are never shared + * @param shareAllQueryStringParameters if true then all query string + * parameters are included + */ + public EvidenceKeyFilterShareUsageTracker( + List blockedHttpHeaders, + List includedQueryStringParams, + boolean includeSession, + String sessionCookieName, + boolean shareAllEvidence, + boolean shareAllQueryStringParameters) { + super( + blockedHttpHeaders, + includedQueryStringParams, + includeSession, + sessionCookieName, + shareAllEvidence, + shareAllQueryStringParameters); + } + @Override public boolean include(String key) { if(key.equalsIgnoreCase(EVIDENCE_SESSIONID) || diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java index 96b24077..2470b8a9 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java @@ -396,6 +396,79 @@ protected ShareUsageBase( List> ignoreDataEvidenceFilter, String sessionCookieName, Tracker tracker) { + this( + logger, + sharePercentage, + minimumEntriesPerMessage, + maximumQueueSize, + addTimeout, + takeTimeout, + repeatEvidenceIntervalMinutes, + trackSession, + shareUsageUrl, + blockedHttpHeaders, + includedQueryStringParameters, + ignoreDataEvidenceFilter, + sessionCookieName, + tracker, + false, + false); + } + + /** + * Constructor + * @param logger the logger to use + * @param sharePercentage the approximate proportion of requests to share. + * 1 = 100%, 0.5 = 50%, etc. + * @param minimumEntriesPerMessage the minimum number of request entries per + * message sent to 51Degrees + * @param maximumQueueSize the maximum number of items to hold in the queue + * at one time. This must be larger than minimum + * entries + * @param addTimeout the timeout in milliseconds to allow when attempting to + * add an item to the queue. If this timeout is exceeded + * then usage sharing will be disabled + * @param takeTimeout the timeout in milliseconds to allow when attempting + * to take an item to the queue + * @param repeatEvidenceIntervalMinutes the interval (in minutes) which is + * used to decide if repeat evidence is + * old enough to consider a new session + * @param trackSession set if the tracker should consider sessions in share + * usage + * @param shareUsageUrl the URL to send data to + * @param blockedHttpHeaders a list of the names of the HTTP headers that + * share usage should not send to 51Degrees + * @param includedQueryStringParameters a list of the names of query string + * parameters that share usage should + * send to 51Degrees + * @param ignoreDataEvidenceFilter the filter used to determine if an item + * of evidence should be ignored or not + * @param sessionCookieName the name of the cookie that contains the session + * id + * @param tracker the {@link Tracker} to use to determine if a given + * {@link FlowData} instance should be shared or not + * @param shareAllEvidence if true then all Evidence is shared, apart from + * the keys that are never shared + * @param shareAllQueryStringParameters if true then all query string + * parameters are shared + */ + protected ShareUsageBase( + Logger logger, + double sharePercentage, + int minimumEntriesPerMessage, + int maximumQueueSize, + int addTimeout, + int takeTimeout, + int repeatEvidenceIntervalMinutes, + boolean trackSession, + String shareUsageUrl, + List blockedHttpHeaders, + List includedQueryStringParameters, + List> ignoreDataEvidenceFilter, + String sessionCookieName, + Tracker tracker, + boolean shareAllEvidence, + boolean shareAllQueryStringParameters) { super(logger, null); if (minimumEntriesPerMessage > maximumQueueSize) { @@ -430,7 +503,12 @@ protected ShareUsageBase( includedQueryStringParameters.add(Constants.EVIDENCE_SEQUENCE_SUFIX); evidenceKeyFilterExclSession = new EvidenceKeyFilterShareUsage( - blockedHttpHeaders, includedQueryStringParameters, false, sessionCookieName); + blockedHttpHeaders, + includedQueryStringParameters, + false, + sessionCookieName, + shareAllEvidence, + shareAllQueryStringParameters); this.ignoreDataEvidenceFilter = ignoreDataEvidenceFilter; @@ -444,7 +522,12 @@ protected ShareUsageBase( cacheConfiguration, intervalMillis, new EvidenceKeyFilterShareUsageTracker( - blockedHttpHeaders, includedQueryStringParameters, trackSession, sessionCookieName)); + blockedHttpHeaders, + includedQueryStringParameters, + trackSession, + sessionCookieName, + shareAllEvidence, + shareAllQueryStringParameters)); } properties = new ArrayList<>(); @@ -711,7 +794,9 @@ private ShareUsageData getDataFromEvidence(Map found) { continue; } - switch (entry.getKey()) { + // The keys are compared without regard to case, and the constants + // are lower case. + switch (entry.getKey().toLowerCase(Locale.ROOT)) { case fiftyone.pipeline.core.Constants.EVIDENCE_CLIENTIP_KEY: // The client IP is dealt with separately for backwards // compatibility purposes. diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilder.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilder.java index 56a96e9a..6e492408 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilder.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilder.java @@ -129,7 +129,9 @@ public ShareUsageElement build() throws IOException { includedQueryStringParameters, ignoreDataEvidenceFilter, sessionCookieName, - null); + null, + shareAllEvidence, + shareAllQueryStringParameters); if (Objects.nonNull(this.httpClient)) { sue.httpClient = httpClient; } else if (Objects.nonNull(dataUploader)){ diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilderBase.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilderBase.java index e376660c..c11954e4 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilderBase.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBuilderBase.java @@ -59,6 +59,8 @@ public abstract class ShareUsageBuilderBase { protected final List includedQueryStringParameters = new ArrayList<>(); protected final List> ignoreDataEvidenceFilter = new ArrayList<>(); protected boolean trackSession; + protected boolean shareAllEvidence; + protected boolean shareAllQueryStringParameters; /** * Constructor @@ -143,6 +145,56 @@ public ShareUsageBuilderBase setIncludedQueryStringParameter( return this; } + /** + * If set to true, every query string and HTTP form parameter is shared, + * rather than only those that start with 51D_ or are named in the + * included parameters. The whole query string and the parameters that are + * never shared are still not shared. + *

+ * Default value is false + * @param shareAll true to share all query string parameters + * @return this builder + */ + @DefaultValue(booleanValue = false) + public ShareUsageBuilderBase setShareAllQueryStringParameters( + boolean shareAll) { + this.shareAllQueryStringParameters = shareAll; + return this; + } + + /** + * Whether every query string parameter is shared. + * @return true if all query string parameters are shared + */ + public boolean isShareAllQueryStringParameters() { + return shareAllQueryStringParameters; + } + + /** + * If set to true, all Evidence is shared, which overrides the blocked + * HTTP headers and the rules for cookies, session values and query string + * parameters. The keys that are never shared (an email address, the salt + * for hashed email identifiers, the whole query string, the path and the + * Cookie header) are still not shared. + *

+ * Default value is false + * @param shareAll true to share all Evidence + * @return this builder + */ + @DefaultValue(booleanValue = false) + public ShareUsageBuilderBase setShareAllEvidence(boolean shareAll) { + this.shareAllEvidence = shareAll; + return this; + } + + /** + * Whether all Evidence is shared. + * @return true if all Evidence is shared + */ + public boolean isShareAllEvidence() { + return shareAllEvidence; + } + /** * By default, all HTTP headers (excluding a few such as 'cookies', if they * don't start with 51D_) are shared. Individual headers can be excluded from sharing by diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java index c56e032f..f680f416 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java @@ -309,6 +309,79 @@ public class ShareUsageElement extends ShareUsageBase { List> ignoreDataEvidenceFilter, String sessionCookieName, Tracker tracker) { + this( + logger, + sharePercentage, + minimumEntriesPerMessage, + maximumQueueSize, + addTimeout, + takeTimeout, + repeatEvidenceIntervalMinutes, + trackSession, + shareUsageUrl, + blockedHttpHeaders, + includedQueryStringParameters, + ignoreDataEvidenceFilter, + sessionCookieName, + tracker, + false, + false); + } + + /** + * Constructor + * @param logger the logger to use + * @param sharePercentage the approximate proportion of requests to share. + * 1 = 100%, 0.5 = 50%, etc. + * @param minimumEntriesPerMessage the minimum number of request entries per + * message sent to 51Degrees + * @param maximumQueueSize the maximum number of items to hold in the queue + * at one time. This must be larger than minimum + * entries + * @param addTimeout the timeout in milliseconds to allow when attempting to + * add an item to the queue. If this timeout is exceeded + * then usage sharing will be disabled + * @param takeTimeout the timeout in milliseconds to allow when attempting + * to take an item to the queue + * @param repeatEvidenceIntervalMinutes the interval (in minutes) which is + * used to decide if repeat evidence is + * old enough to consider a new session + * @param trackSession set if the tracker should consider sessions in share + * usage + * @param shareUsageUrl the URL to send data to + * @param blockedHttpHeaders a list of the names of the HTTP headers that + * share usage should not send to 51Degrees + * @param includedQueryStringParameters a list of the names of query string + * parameters that share usage should + * send to 51Degrees + * @param ignoreDataEvidenceFilter the filter used to determine if an item + * of evidence should be ignored or not + * @param sessionCookieName the name of the cookie that contains the session + * id + * @param tracker the {@link Tracker} to use to determine if a given + * {@link FlowData} instance should be shared or not + * @param shareAllEvidence if true then all Evidence is shared, apart from + * the keys that are never shared + * @param shareAllQueryStringParameters if true then all query string + * parameters are shared + */ + ShareUsageElement( + Logger logger, + double sharePercentage, + int minimumEntriesPerMessage, + int maximumQueueSize, + int addTimeout, + int takeTimeout, + int repeatEvidenceIntervalMinutes, + boolean trackSession, + String shareUsageUrl, + List blockedHttpHeaders, + List includedQueryStringParameters, + List> ignoreDataEvidenceFilter, + String sessionCookieName, + Tracker tracker, + boolean shareAllEvidence, + boolean shareAllQueryStringParameters) { super( logger, sharePercentage, @@ -323,7 +396,9 @@ public class ShareUsageElement extends ShareUsageBase { includedQueryStringParameters, ignoreDataEvidenceFilter, sessionCookieName, - tracker); + tracker, + shareAllEvidence, + shareAllQueryStringParameters); headers.put("Content-Type", "text/xml; charset=utf-8"); // assume default GZIP connection diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/trackers/ShareUsageTracker.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/trackers/ShareUsageTracker.java index c6082300..5c8f2e47 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/trackers/ShareUsageTracker.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/trackers/ShareUsageTracker.java @@ -22,7 +22,6 @@ package fiftyone.pipeline.engines.fiftyone.trackers; -import fiftyone.caching.PutCache; import fiftyone.pipeline.core.data.DataKey; import fiftyone.pipeline.core.data.DataKeyBuilder; import fiftyone.pipeline.core.data.DataKeyBuilderDefault; @@ -32,7 +31,6 @@ import fiftyone.pipeline.engines.fiftyone.data.ShareUsageEvidence; import fiftyone.pipeline.engines.trackers.TrackerBase; -import java.io.IOException; import java.util.Date; import java.util.Map; @@ -53,13 +51,6 @@ public class ShareUsageTracker extends TrackerBase { */ private final long interval; - /** - * The times keyed by the values usage sharing would share. The cache of - * the base class is keyed by the Evidence alone, which no longer holds - * the web request's values. - */ - private final PutCache sharedValuesCache; - /** * Construct a new instance. * @param configuration the cache configuration to use when building the @@ -75,10 +66,6 @@ public ShareUsageTracker( super(configuration); this.interval = intervalMillis; this.filter = filter; - // The base class has checked that the builder makes a PutCache. - this.sharedValuesCache = (PutCache) - configuration.getCacheBuilder().build( - null, configuration.getSize()); } @Override @@ -98,31 +85,14 @@ protected EvidenceKeyFilter getFilter() { return filter; } - @Override - public Date get(FlowData data) { - return sharedValuesCache.get(getSharedValuesKey(data)); - } - - @Override - public void put(FlowData data, Date value) { - sharedValuesCache.put(getSharedValuesKey(data), value); - } - - @Override - public void close() throws IOException { - try { - sharedValuesCache.close(); - } finally { - super.close(); - } - } - /** - * The key of the values {@link ShareUsageEvidence#find} finds that the - * filter includes, which take in the web request's values, as the - * session tracking hash is of the values that would be shared. + * The key is made from the values {@link ShareUsageEvidence#find} finds + * that the filter includes, which take in the web request's values, + * as the session tracking hash is of the values that would be shared. + * The Evidence alone does not hold them. */ - private DataKey getSharedValuesKey(FlowData data) { + @Override + protected DataKey generateKey(FlowData data) { DataKeyBuilder builder = new DataKeyBuilderDefault(); for (Map.Entry entry : ShareUsageEvidence.find(data.getEvidence()).entrySet()) { diff --git a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageOptionsTests.java b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageOptionsTests.java index ef2e2610..65a92048 100644 --- a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageOptionsTests.java +++ b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageOptionsTests.java @@ -236,6 +236,15 @@ public void shareAllQueryStringParameters_False_FromConfiguration_KeepsTheRules( assertFalse(shared(sent, "query", "plain")); } + @Test + public void builder_ShareOptionsDefaultToFalse() { + ShareUsageBuilder builder = new ShareUsageBuilder( + LoggerFactory.getILoggerFactory()); + + assertFalse(builder.isShareAllEvidence()); + assertFalse(builder.isShareAllQueryStringParameters()); + } + @Test public void keysAreReadWithoutRegardToCase() throws Exception { Map evidence = new HashMap<>(); diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/caching/DataKeyedCacheBase.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/caching/DataKeyedCacheBase.java index fd16b2c1..348f613a 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/caching/DataKeyedCacheBase.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/caching/DataKeyedCacheBase.java @@ -70,14 +70,26 @@ public DataKeyedCacheBase(CacheConfiguration configuration) { */ protected abstract EvidenceKeyFilter getFilter(); + /** + * Creates the key the internal cache holds the data of a {@link FlowData} + * under. By default this is the key the {@link FlowData} generates from + * the Evidence that {@link #getFilter()} includes. A class whose key is + * made from other values overrides this method. + * @param flowData the {@link FlowData} to make a key for + * @return the key to use in the internal cache + */ + protected DataKey generateKey(FlowData flowData) { + return flowData.generateKey(getFilter()); + } + @Override public V get(FlowData flowData) { - return internalCache.get(flowData.generateKey(getFilter())); + return internalCache.get(generateKey(flowData)); } @Override public void put(FlowData flowData, V value) { - internalCache.put(flowData.generateKey(getFilter()), value); + internalCache.put(generateKey(flowData), value); } @Override From 8a99122528b18e1697858c0d748615dcd0871cc0 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 21:35:06 +0000 Subject: [PATCH 07/16] Test the timeouts on the legacy send --- .../ShareUsageSendTimeoutTests.java | 133 ++++++++++++++++++ 1 file changed, 133 insertions(+) create mode 100644 pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageSendTimeoutTests.java diff --git a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageSendTimeoutTests.java b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageSendTimeoutTests.java new file mode 100644 index 00000000..780cb80b --- /dev/null +++ b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageSendTimeoutTests.java @@ -0,0 +1,133 @@ +/* ********************************************************************* + * This Original Work is copyright of 51 Degrees Mobile Experts Limited. + * Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, + * Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. + * + * This Original Work is licensed under the European Union Public Licence + * (EUPL) v.1.2 and is subject to its terms as set out below. + * + * If a copy of the EUPL was not distributed with this file, You can obtain + * one at https://opensource.org/licenses/EUPL-1.2. + * + * The 'Compatible Licences' set out in the Appendix to the EUPL (as may be + * amended by the European Commission) shall be deemed incompatible for + * the purposes of the Work and the provisions of the compatibility + * clause in Article 5 of the EUPL shall not apply. + * + * If using the Work as, or as part of, a network application, by + * including the attribution notice(s) required under Article 5 of the EUPL + * in the end user terms of the application under an appropriate heading, + * such notice(s) shall fulfill the requirements of that article. + * ********************************************************************* */ + +package fiftyone.pipeline.engines.fiftyone.flowelements; + +import fiftyone.pipeline.engines.services.HttpClient; +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; + +import java.net.HttpURLConnection; +import java.net.URL; +import java.util.ArrayList; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * A send through an {@link HttpClient} that never answers must end, or it + * would stop all sending. + */ +public class ShareUsageSendTimeoutTests { + + /** + * Opens a connection that records the timeouts set on it. + */ + private static class RecordingHttpClient implements HttpClient { + private HttpURLConnection connection; + + @Override + public HttpURLConnection connect(URL url) { + connection = new HttpURLConnection(url) { + @Override + public int getResponseCode() { + return 200; + } + + @Override + public String getResponseMessage() { + return ""; + } + + @Override + public void disconnect() { } + + @Override + public boolean usingProxy() { + return false; + } + + @Override + public void connect() { } + }; + return connection; + } + + @Override + public String postData( + HttpURLConnection connection, + Map headers, + byte[] data) { + return ""; + } + + @Override + public String getResponseString(HttpURLConnection connection) { + return ""; + } + + @Override + public String getResponseString( + HttpURLConnection connection, + Map headers) { + return ""; + } + } + + @Test + public void ShareUsageElement_LegacySend_ConnectionHasTimeouts() + throws Exception { + RecordingHttpClient client = new RecordingHttpClient(); + ShareUsageElement element = new ShareUsageElement( + LoggerFactory.getLogger(ShareUsageElement.class), + 1, + 1, + 100, + 100, + 100, + 1, + true, + "http://localhost/usage", + new ArrayList<>(), + new ArrayList<>(), + new ArrayList<>(), + fiftyone.pipeline.engines.Constants.DEFAULT_SESSION_COOKIE_NAME, + flowData -> true); + element.httpClient = client; + try { + element.legacySendAsXML(new ArrayList<>()); + + assertEquals( + element.httpSendTimeout, + client.connection.getConnectTimeout(), + "The connection has no connect timeout."); + assertEquals( + element.httpSendTimeout, + client.connection.getReadTimeout(), + "The connection has no read timeout."); + assertTrue(element.httpSendTimeout > 0); + } finally { + element.close(); + } + } +} From 4e3838d8b45a6d59b56b68be08761298989ab93e Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 21:35:15 +0000 Subject: [PATCH 08/16] Test the back-off time comes from the monotonic clock --- .../flowelements/ShareUsageBackOffTests.java | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOffTests.java b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOffTests.java index b9716f9f..1a628235 100644 --- a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOffTests.java +++ b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOffTests.java @@ -62,6 +62,19 @@ private static ShareUsageBackOff create() { return create(3, 2, 5); } + /** + * The time the back-off is given comes from a clock that a change to the + * system time does not move. + */ + @Test + public void ShareUsageBackOff_Now_FollowsTheMonotonicClock() { + long before = System.nanoTime() / 1000000; + long now = ShareUsageBackOff.now(); + long after = System.nanoTime() / 1000000; + + assertTrue(before <= now && now <= after); + } + @Test public void ShareUsageBackOff_PausesAfterFailuresWithinWindow() { ShareUsageBackOff backOff = create(); From a801f8bb976b0efe2e782ec3df31ab32131970a8 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 21:36:06 +0000 Subject: [PATCH 09/16] Time pauses on a monotonic clock and time out the legacy send --- .../fiftyone/flowelements/ShareUsageBackOff.java | 12 +++++++++++- .../fiftyone/flowelements/ShareUsageBase.java | 6 +++--- .../fiftyone/flowelements/ShareUsageElement.java | 3 +++ 3 files changed, 17 insertions(+), 4 deletions(-) diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java index 8cc06753..2a582d7b 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBackOff.java @@ -30,7 +30,9 @@ * once the set number of sends fail within the window. A failed send after a * pause multiplies the delay, up to the maximum, and a successful send ends * the back-off. Only sending pauses, so evidence is still queued while - * paused. Each method takes the time from the caller, in milliseconds. + * paused. Each method takes the time from the caller, in milliseconds on + * the clock that {@link #now()} reads, so a change to the system time cannot + * lengthen or skip a pause. */ class ShareUsageBackOff { @@ -57,6 +59,14 @@ class ShareUsageBackOff { */ private long resumeAt = Long.MIN_VALUE; + /** + * @return the time in milliseconds on a monotonic clock, for the caller to + * pass to the other methods. Its zero is not a calendar time. + */ + static long now() { + return System.nanoTime() / 1000000; + } + ShareUsageBackOff( int failuresToEnterRecovery, int failuresWindowSeconds, diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java index d74029b2..e42d6bab 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java @@ -750,7 +750,7 @@ protected void managedResourcesCleanup() { logger.warn("Interrupted while awaiting close down"); } if (evidenceCollection.isEmpty() == false && - backOff.remaining(System.currentTimeMillis()) > 0) { + backOff.remaining(ShareUsageBackOff.now()) > 0) { logger.info( "{} usage sharing entries were not sent, as sending was " + "paused after failed sends when usage sharing closed.", @@ -925,7 +925,7 @@ private void submitSendData() { */ protected boolean waitWhilePaused() { long pause; - while ((pause = backOff.remaining(System.currentTimeMillis())) > 0) { + while ((pause = backOff.remaining(ShareUsageBackOff.now())) > 0) { try { if (closing.await(pause, TimeUnit.MILLISECONDS)) { return false; @@ -950,7 +950,7 @@ protected void recordSendResult(boolean accepted) { logger.info("Usage sharing resumed after a send succeeded."); } } else { - long pause = backOff.failed(System.currentTimeMillis()); + long pause = backOff.failed(ShareUsageBackOff.now()); if (pause > 0) { logger.warn( "Usage sharing paused for {} seconds after failed " + diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java index 8033b956..6cfe7b91 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java @@ -670,6 +670,9 @@ protected void legacySendAsXML(List allData) throws Exception { } HttpURLConnection connection = httpClient.connect(new URI(shareUsageUrl.trim()).toURL()); + // A send that is never answered must end, or it stops all sending. + connection.setConnectTimeout(httpSendTimeout); + connection.setReadTimeout(httpSendTimeout); String response = httpClient.postData(connection, headers, baos.toByteArray()); int responseCode = connection.getResponseCode(); String responseMessage = connection.getResponseMessage() + "data: '" + response + "'"; From cd541e45a6ba6a3a3ea90034453fbb2ad6e0362e Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 21:37:11 +0000 Subject: [PATCH 10/16] Test that an upload to a service that never answers times out --- .../services/DataUploaderHttpTests.java | 84 +++++++++++++++++++ 1 file changed, 84 insertions(+) create mode 100644 pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUploaderHttpTests.java diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUploaderHttpTests.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUploaderHttpTests.java new file mode 100644 index 00000000..2954f257 --- /dev/null +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUploaderHttpTests.java @@ -0,0 +1,84 @@ +/* ********************************************************************* + * This Original Work is copyright of 51 Degrees Mobile Experts Limited. + * Copyright 2026 51 Degrees Mobile Experts Limited, Davidson House, + * Forbury Square, Reading, Berkshire, United Kingdom RG1 3EU. + * + * This Original Work is licensed under the European Union Public Licence + * (EUPL) v.1.2 and is subject to its terms as set out below. + * + * If a copy of the EUPL was not distributed with this file, You can obtain + * one at https://opensource.org/licenses/EUPL-1.2. + * + * The 'Compatible Licences' set out in the Appendix to the EUPL (as may be + * amended by the European Commission) shall be deemed incompatible for + * the purposes of the Work and the provisions of the compatibility + * clause in Article 5 of the EUPL shall not apply. + * + * If using the Work as, or as part of, a network application, by + * including the attribution notice(s) required under Article 5 of the EUPL + * in the end user terms of the application under an appropriate heading, + * such notice(s) shall fulfill the requirements of that article. + * ********************************************************************* */ + +package fiftyone.pipeline.engines.services; + +import static org.junit.Assert.fail; + +import java.io.OutputStream; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.util.HashMap; + +import org.junit.Test; + +/** + * An upload to a service that accepts the connection and never answers must + * end, or it would stop all usage sharing. + */ +public class DataUploaderHttpTests { + + /** + * How long the uploader waits for an answer. + */ + private static final int TIMEOUT_MILLIS = 500; + + /** + * How long the test service holds a connection open without answering. + */ + private static final int HOLD_MILLIS = 30000; + + /** + * The test fails if the upload has not ended by then. + */ + private static final int LIMIT_MILLIS = 15000; + + @Test(timeout = LIMIT_MILLIS) + public void DataUploaderHttp_NoAnswer_TimesOut() throws Exception { + try (ServerSocket server = new ServerSocket(0)) { + Thread holder = new Thread(() -> { + try (Socket socket = server.accept()) { + Thread.sleep(HOLD_MILLIS); + } catch (Exception ignored) { + // The test has ended and closed the server. + } + }); + holder.setDaemon(true); + holder.start(); + DataUploaderHttp uploader = new DataUploaderHttp( + "http://localhost:" + server.getLocalPort() + "/usage", + new HashMap<>(), + TIMEOUT_MILLIS); + try (OutputStream out = uploader.getOutputStream()) { + out.write(new byte[] {1}); + } + + try { + uploader.getResponseCode(); + fail("The upload was answered."); + } catch (SocketTimeoutException expected) { + // The wait for an answer ended. + } + } + } +} From 9eca2e4bbc32ceaf4f020fac1c496595a88ac4fb Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 21:37:39 +0000 Subject: [PATCH 11/16] Time out an upload that is never answered --- .../fiftyone/pipeline/engines/services/DataUploaderHttp.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploaderHttp.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploaderHttp.java index 777584bf..0c2b044d 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploaderHttp.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploaderHttp.java @@ -40,7 +40,8 @@ public class DataUploaderHttp implements DataUploader{ * the URL to which data is to be uploaded * @param url the URL rto which data should be uploaded * @param headers HTTP headers for the connection - * @param timeout connection timeout in milliseconds + * @param timeout milliseconds to wait to connect, and then to wait for + * an answer */ public DataUploaderHttp(String url, Map headers, int timeout) { this.url = url; @@ -55,6 +56,7 @@ public DataUploaderHttp(String url, Map headers, int timeout) { public OutputStream getOutputStream() throws Exception{ connection = (HttpURLConnection) new URI(url.trim()).toURL().openConnection(); connection.setConnectTimeout(timeout); + connection.setReadTimeout(timeout); connection.setRequestMethod("POST"); for (Map.Entry header : headers.entrySet()) { connection.setRequestProperty(header.getKey(), header.getValue()); From 80729bb61f650ffea78724a6328c7ebefeabe64b Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 21:58:36 +0000 Subject: [PATCH 12/16] Test that discarding entries from a full queue is logged without a message per entry --- .../ShareUsageElementBackOffTests.java | 56 ++++++++++++++++++- 1 file changed, 55 insertions(+), 1 deletion(-) diff --git a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java index 41daf584..b125eea4 100644 --- a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java +++ b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java @@ -30,14 +30,18 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.ByteArrayOutputStream; import java.io.OutputStream; +import java.lang.reflect.Proxy; import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.Arrays; import java.util.HashMap; import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.Map; import java.util.function.BooleanSupplier; import java.util.function.Supplier; @@ -129,6 +133,7 @@ public boolean track(FlowData flowData) { } private Pipeline pipeline; + private Logger logger = LoggerFactory.getLogger(ShareUsageElement.class); private RecordingUploader uploader; private ShareUsageElement element; private int requests; @@ -172,7 +177,7 @@ private void useBackOff( int maximumQueueSize) throws Exception { element.close(); element = new ShareUsageElement( - LoggerFactory.getLogger(ShareUsageElement.class), + logger, 1, 1, maximumQueueSize, @@ -357,4 +362,53 @@ public void ShareUsageElement_BackOff_FullQueueDropsNewEvidence() "Evidence processed when the queue was full was sent."); } } + + /** + * A logger that records the warnings it is given. + */ + private static Logger recordingLogger(List warnings) { + return (Logger) Proxy.newProxyInstance( + Logger.class.getClassLoader(), + new Class[] {Logger.class}, + (proxy, method, args) -> { + if (method.getName().equals("warn")) { + warnings.add(Arrays.deepToString(args)); + return null; + } + if (method.getReturnType() == boolean.class) { + return false; + } + if (method.getReturnType() == int.class) { + return 0; + } + if (method.getReturnType() == String.class) { + return "test"; + } + return null; + }); + } + + /** + * Evidence discarded because the queue is full is logged, but not with a + * message for each entry discarded. + */ + @Test + public void ShareUsageElement_FullQueue_LogsWithoutAMessagePerEntry() + throws Exception { + List warnings = new CopyOnWriteArrayList<>(); + logger = recordingLogger(warnings); + useBackOff(1, 2, 2, 4); + failSends(1); + warnings.clear(); + + for (int i = 0; i < 10; i++) { + process(); + } + + long full = warnings.stream() + .filter(warning -> warning.contains("full")) + .count(); + assertTrue(full >= 1, "Discarding entries was not logged."); + assertTrue(full < 6, "A message was logged for each entry discarded."); + } } From 2ea0043cd978e3ff74e2c71070d3e9cc4297eb77 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 21:58:55 +0000 Subject: [PATCH 13/16] Log entries discarded from a full queue without a message per entry --- .../fiftyone/flowelements/Constants.java | 7 ++++++ .../fiftyone/flowelements/ShareUsageBase.java | 24 +++++++++++++++---- 2 files changed, 26 insertions(+), 5 deletions(-) diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/Constants.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/Constants.java index f2e536e8..28506e77 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/Constants.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/Constants.java @@ -93,6 +93,13 @@ public class Constants { * Default lost data count for share usage before it wraps */ static final int LOST_DATA_RESET_DEFAULT = 5000000; + + /** + * While the queue is full, a message is logged for the first entry + * discarded and then once for every this many more, never one for each + * entry. + */ + static final int QUEUE_FULL_LOG_EVERY = 1000; /** * The default URL to send usage data to. */ diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java index e42d6bab..9e417dcf 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageBase.java @@ -228,6 +228,13 @@ public abstract class ShareUsageBase */ protected long lostDataReset = LOST_DATA_RESET_DEFAULT; + /** + * True from the first entry discarded because the queue is full until an + * entry is next queued. While true, a message is logged only for every + * {@link Constants#QUEUE_FULL_LOG_EVERY} entries discarded. + */ + private volatile boolean queueFullReported = false; + /** * Decides when sending may happen after failed sends. */ @@ -796,6 +803,7 @@ private void processData(FlowData data, Map found) { getDataFromEvidence(found), addTimeout, TimeUnit.MILLISECONDS)) { + queueFullReported = false; logger.debug("Queued {} entries", evidenceCollection.size()); // If the collection has enough entries then start // taking data from it to be sent. @@ -803,11 +811,17 @@ private void processData(FlowData data, Map found) { trySendData(); } } else { - if (++lostData >= lostDataReset) { - logger.warn("Could not queue data for Share Usage. This could mean that" + - " the queue is too small, that the proportion of data shared is" + - " too high - or that the sending process has encountered" + - " problems sending."); + long lost = ++lostData; + if (queueFullReported == false || + lost % Constants.QUEUE_FULL_LOG_EVERY == 0) { + queueFullReported = true; + logger.warn("The usage sharing queue is full, so usage data is " + + "being discarded. This could mean that the queue is too " + + "small, that the proportion of data shared is too high, " + + "or that sending is failing. {} entries have been " + + "discarded.", lost); + } + if (lost >= lostDataReset) { lostData = 0; } } From 680ba4e1c3cca0d155163e2bbad1faac7acc146e Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 21:59:18 +0000 Subject: [PATCH 14/16] Test that a failed answer reports the status and the content --- .../ShareUsageElementBackOffTests.java | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java index b125eea4..0a640555 100644 --- a/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java +++ b/pipeline/pipeline.engines.fiftyone/src/test/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElementBackOffTests.java @@ -51,6 +51,7 @@ import static fiftyone.pipeline.core.Constants.EVIDENCE_CLIENTIP_KEY; import static fiftyone.pipeline.engines.Constants.DEFAULT_SESSION_COOKIE_NAME; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; @@ -83,6 +84,7 @@ private static class RecordingUploader implements DataUploader { private final List times = new ArrayList<>(); private final List accepted = new ArrayList<>(); private volatile int code = 500; + private volatile String content = ""; private ByteArrayOutputStream body; @Override @@ -105,6 +107,10 @@ public synchronized int getResponseCode() { return answer; } + public String getResponseContent() { + return content; + } + long now() { return (System.nanoTime() - start) / 1000000; } @@ -411,4 +417,20 @@ public void ShareUsageElement_FullQueue_LogsWithoutAMessagePerEntry() assertTrue(full >= 1, "Discarding entries was not logged."); assertTrue(full < 6, "A message was logged for each entry discarded."); } + + /** + * A send answered with anything other than 200 reports the status code + * and the content of the answer. + */ + @Test + public void ShareUsageElement_FailedAnswer_ReportsStatusAndContent() { + uploader.content = "the service says no"; + + Exception failure = assertThrows( + Exception.class, + () -> element.sendAsXML(new ArrayList<>())); + + assertTrue(failure.getMessage().contains("500")); + assertTrue(failure.getMessage().contains("the service says no")); + } } From b81c1a1615018ec87a0d2e6721861fc0bbbe22c8 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 21:59:36 +0000 Subject: [PATCH 15/16] Test that a failed upload gives the content of the answer --- .../services/DataUploaderHttpTests.java | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUploaderHttpTests.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUploaderHttpTests.java index 2954f257..5948dfac 100644 --- a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUploaderHttpTests.java +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUploaderHttpTests.java @@ -22,9 +22,14 @@ package fiftyone.pipeline.engines.services; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.fail; +import com.sun.net.httpserver.HttpServer; + import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; import java.net.ServerSocket; import java.net.Socket; import java.net.SocketTimeoutException; @@ -81,4 +86,33 @@ public void DataUploaderHttp_NoAnswer_TimesOut() throws Exception { } } } + + @Test(timeout = LIMIT_MILLIS) + public void DataUploaderHttp_FailedAnswer_GivesTheContent() + throws Exception { + byte[] answer = "the service says no".getBytes(StandardCharsets.UTF_8); + HttpServer server = HttpServer.create(new InetSocketAddress(0), 0); + server.createContext("/usage", exchange -> { + exchange.sendResponseHeaders(500, answer.length); + exchange.getResponseBody().write(answer); + exchange.close(); + }); + server.start(); + try { + DataUploaderHttp uploader = new DataUploaderHttp( + "http://localhost:" + server.getAddress().getPort() + "/usage", + new HashMap<>(), + TIMEOUT_MILLIS * 10); + try (OutputStream out = uploader.getOutputStream()) { + out.write(new byte[] {1}); + } + + assertEquals(500, uploader.getResponseCode()); + assertEquals( + "the service says no", + uploader.getResponseContent()); + } finally { + server.stop(0); + } + } } From aabb62dca2f85889156311d6aa9453e241c5e48f Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Mon, 9 Nov 2026 21:59:51 +0000 Subject: [PATCH 16/16] Report the status and the content of a failed answer --- .../flowelements/ShareUsageElement.java | 5 +++- .../engines/services/DataUploader.java | 7 +++++ .../engines/services/DataUploaderHttp.java | 30 +++++++++++++++++++ 3 files changed, 41 insertions(+), 1 deletion(-) diff --git a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java index 6cfe7b91..37a5eba7 100644 --- a/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java +++ b/pipeline/pipeline.engines.fiftyone/src/main/java/fiftyone/pipeline/engines/fiftyone/flowelements/ShareUsageElement.java @@ -509,7 +509,10 @@ protected void sendAsXML(List allData) throws Exception { // check for completion int code = dataUploader.getResponseCode(); if (code != 200) { - throw new Exception(String.format("Share Usage response code was %d ", code)); + throw new Exception(String.format( + "Share Usage response code was %d with content '%s'", + code, + dataUploader.getResponseContent())); } logger.debug(threadMarker, "Send elements done"); } diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploader.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploader.java index 38df1b00..b790a641 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploader.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploader.java @@ -44,5 +44,12 @@ public interface DataUploader { */ int getResponseCode() throws IOException; + /** + * @return the content of the answer, which is logged when the code is + * not 200, or an empty string if the uploader cannot give it + */ + default String getResponseContent() { + return ""; + } } diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploaderHttp.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploaderHttp.java index 0c2b044d..9ad54e67 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploaderHttp.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUploaderHttp.java @@ -23,9 +23,11 @@ package fiftyone.pipeline.engines.services; import java.io.IOException; +import java.io.InputStream; import java.io.OutputStream; import java.net.HttpURLConnection; import java.net.URI; +import java.nio.charset.StandardCharsets; import java.util.Map; import java.util.zip.GZIPOutputStream; @@ -34,6 +36,11 @@ public class DataUploaderHttp implements DataUploader{ private final String url; private final Map headers; private final int timeout; + + /** + * The most characters of an answer's content that are given. + */ + private static final int LOGGED_CONTENT_LENGTH = 500; HttpURLConnection connection; /** @@ -69,5 +76,28 @@ public OutputStream getOutputStream() throws Exception{ public int getResponseCode() throws IOException { return connection.getResponseCode(); } + + /** + * Reads the content of a failed answer, as far as the most characters + * that are logged. + */ + @Override + public String getResponseContent() { + try (InputStream error = connection.getErrorStream()) { + if (error == null) { + return ""; + } + byte[] content = new byte[LOGGED_CONTENT_LENGTH]; + int length = 0; + int read; + while (length < content.length && (read = error.read( + content, length, content.length - length)) > 0) { + length += read; + } + return new String(content, 0, length, StandardCharsets.UTF_8); + } catch (IOException e) { + return ""; + } + } }