|
5 | 5 |
|
6 | 6 | import com.azure.core.util.logging.ClientLogger; |
7 | 7 | import com.azure.messaging.servicebus.ServiceBusProcessorClient; |
| 8 | +import com.azure.messaging.servicebus.ServiceBusReceivedMessage; |
8 | 9 | import com.azure.messaging.servicebus.ServiceBusReceivedMessageContext; |
| 10 | +import com.azure.messaging.servicebus.stress.util.RunResult; |
9 | 11 | import org.springframework.beans.factory.annotation.Value; |
10 | 12 | import org.springframework.stereotype.Component; |
11 | 13 |
|
12 | 14 | import java.time.Duration; |
| 15 | +import java.time.OffsetDateTime; |
13 | 16 | import java.util.concurrent.ThreadLocalRandom; |
| 17 | +import java.util.concurrent.atomic.AtomicReference; |
14 | 18 |
|
15 | 19 | import static com.azure.messaging.servicebus.stress.scenarios.TestUtils.blockingWait; |
| 20 | +import static com.azure.messaging.servicebus.stress.scenarios.TestUtils.createMessagePayload; |
16 | 21 | import static com.azure.messaging.servicebus.stress.scenarios.TestUtils.getProcessorBuilder; |
17 | 22 |
|
18 | 23 | /** |
|
22 | 27 | public class MessageProcessor extends ServiceBusScenario { |
23 | 28 | private static final ClientLogger LOGGER = new ClientLogger(MessageProcessor.class); |
24 | 29 |
|
25 | | - @Value("${DURATION_IN_MINUTES:15}") |
26 | | - private int testDurationInMinutes; |
27 | | - |
28 | | - // lock duration is 5 sec, so in some cases we'll do lock renewal |
29 | | - @Value("${PROCESS_CALLBACK_DURATION_MAX_IN_SECONDS:7}") |
30 | | - private int processMessageDurationMaxInSeconds; |
| 30 | + @Value("${PROCESS_CALLBACK_DURATION_MAX_IN_MS:50}") |
| 31 | + private int processMessageDurationMaxInMs; |
31 | 32 |
|
32 | 33 | @Value("${MAX_CONCURRENT_CALLS:100}") |
33 | 34 | private int maxConcurrentCalls; |
34 | 35 |
|
35 | 36 | @Value("${PREFETCH_COUNT:0}") |
36 | 37 | private int prefetchCount; |
37 | 38 |
|
| 39 | + @Value("${ABANDON_RATIO:0}") |
| 40 | + private double abandonRatio; |
| 41 | + |
| 42 | + @Value("${NO_DISPOSITION_RATIO:0}") |
| 43 | + private double noDispositionRatio; |
| 44 | + |
| 45 | + @Value("${LOCK_RENEWAL_NEEDED_RATIO:0}") |
| 46 | + private double lockRenewalNeededRatio; |
| 47 | + |
| 48 | + @Value("${LOCK_DURATION_IN_MS:30000}") |
| 49 | + private int lockDurationInMs; |
| 50 | + |
| 51 | + @Value("${AUTO_RENEW_LOCK:true}") |
| 52 | + private boolean renewLock; |
| 53 | + |
| 54 | + private byte[] expectedPayload; |
| 55 | + |
| 56 | + private final AtomicReference<RunResult> runResult = new AtomicReference<>(RunResult.INCONCLUSIVE); |
| 57 | + |
| 58 | + |
38 | 59 | @Override |
39 | | - public void run() { |
40 | | - ServiceBusProcessorClient processor = getProcessorBuilder(options) |
41 | | - .maxAutoLockRenewDuration(Duration.ofSeconds(processMessageDurationMaxInSeconds + 1)) |
| 60 | + public RunResult run() throws InterruptedException { |
| 61 | + expectedPayload = createMessagePayload(options.getMessageSize()); |
| 62 | + |
| 63 | + ServiceBusProcessorClient processor = toClose(getProcessorBuilder(options) |
| 64 | + .maxAutoLockRenewDuration(renewLock ? Duration.ofMinutes(5) : Duration.ZERO) |
42 | 65 | .maxConcurrentCalls(maxConcurrentCalls) |
43 | 66 | .prefetchCount(prefetchCount) |
44 | 67 | .processMessage(this::process) |
45 | 68 | .processError(err -> { |
46 | | - throw LOGGER.logExceptionAsError(new RuntimeException(err.getException())); |
| 69 | + LOGGER.atError() |
| 70 | + .addKeyValue("source", err.getErrorSource()) |
| 71 | + .log("processor error", err.getException()); |
| 72 | + runResult.set(RunResult.ERROR); |
47 | 73 | }) |
48 | | - .buildProcessorClient(); |
49 | | - |
| 74 | + .buildProcessorClient()); |
50 | 75 | processor.start(); |
51 | | - blockingWait(Duration.ofMinutes(testDurationInMinutes)); |
52 | | - processor.close(); |
| 76 | + |
| 77 | + blockingWait(options.getTestDuration()); |
| 78 | + |
| 79 | + int activeMessages = getRemainingQueueMessages(); |
| 80 | + for (int extraMinutes = 0; extraMinutes < 3 && activeMessages > 0; extraMinutes++) { |
| 81 | + blockingWait(Duration.ofMinutes(1)); |
| 82 | + activeMessages = getRemainingQueueMessages(); |
| 83 | + } |
| 84 | + |
| 85 | + return activeMessages != 0 ? RunResult.WARNING : runResult.get(); |
53 | 86 | } |
54 | 87 |
|
55 | 88 | private void process(ServiceBusReceivedMessageContext messageContext) { |
| 89 | + ServiceBusReceivedMessage message = messageContext.getMessage(); |
| 90 | + if (checkMessage(message)) { |
| 91 | + blockingWait(Duration.ofMillis(getWaitTime())); |
| 92 | + settleMessage(messageContext); |
| 93 | + } |
| 94 | + } |
| 95 | + |
| 96 | + private int getWaitTime() { |
| 97 | + ThreadLocalRandom random = ThreadLocalRandom.current(); |
| 98 | + if (random.nextDouble(1) < lockRenewalNeededRatio) { |
| 99 | + return lockDurationInMs + 1000; |
| 100 | + } else if (processMessageDurationMaxInMs != 0) { |
| 101 | + return random.nextInt(processMessageDurationMaxInMs); |
| 102 | + } |
| 103 | + |
| 104 | + return 0; |
| 105 | + } |
| 106 | + |
| 107 | + private boolean checkMessage(ServiceBusReceivedMessage message) { |
| 108 | + LOGGER.atInfo() |
| 109 | + .addKeyValue("messageId", message.getMessageId()) |
| 110 | + .addKeyValue("traceparent", message.getApplicationProperties().get("traceparent")) |
| 111 | + .addKeyValue("deliveryCount", message.getDeliveryCount()) |
| 112 | + .addKeyValue("lockToken", message.getLockToken()) |
| 113 | + .addKeyValue("lockedUntil", message.getLockedUntil()) |
| 114 | + .log("message received"); |
| 115 | + |
| 116 | + if (message.getLockedUntil().isBefore(OffsetDateTime.now())) { |
| 117 | + LOGGER.atError() |
| 118 | + .addKeyValue("messageId", message.getMessageId()) |
| 119 | + .addKeyValue("deliveryCount", message.getDeliveryCount()) |
| 120 | + .log("message lock expired"); |
| 121 | + runResult.set(RunResult.ERROR); |
| 122 | + return false; |
| 123 | + } |
| 124 | + |
| 125 | + byte[] payload = message.getBody().toBytes(); |
| 126 | + if (payload.length != expectedPayload.length) { |
| 127 | + LOGGER.atError() |
| 128 | + .addKeyValue("messageId", message.getMessageId()) |
| 129 | + .addKeyValue("actualSize", payload.length) |
| 130 | + .addKeyValue("expectedSize", expectedPayload.length) |
| 131 | + .log("message corrupted"); |
| 132 | + runResult.set(RunResult.ERROR); |
| 133 | + } |
| 134 | + |
| 135 | + for (int i = 0; i < payload.length; i++) { |
| 136 | + if (payload[i] != expectedPayload[i]) { |
| 137 | + LOGGER.atError() |
| 138 | + .addKeyValue("messageId", message.getMessageId()) |
| 139 | + .addKeyValue("index", i) |
| 140 | + .addKeyValue("actual", payload[i]) |
| 141 | + .addKeyValue("expected", expectedPayload[i]) |
| 142 | + .log("message corrupted"); |
| 143 | + runResult.set(RunResult.ERROR); |
| 144 | + } |
| 145 | + } |
| 146 | + |
| 147 | + return true; |
| 148 | + } |
| 149 | + |
| 150 | + private void settleMessage(ServiceBusReceivedMessageContext messageContext) { |
| 151 | + String operation = "ignored"; |
56 | 152 | try { |
57 | | - if (processMessageDurationMaxInSeconds != 0) { |
58 | | - int processTimeMs = ThreadLocalRandom.current().nextInt(processMessageDurationMaxInSeconds * 1000); |
59 | | - Thread.sleep(processTimeMs); |
| 153 | + double random = ThreadLocalRandom.current().nextDouble(1); |
| 154 | + if (random < abandonRatio) { |
| 155 | + operation = "abandoned"; |
| 156 | + messageContext.abandon(); |
| 157 | + } else if (random >= abandonRatio + noDispositionRatio) { |
| 158 | + operation = "completed"; |
| 159 | + messageContext.complete(); |
60 | 160 | } |
61 | | - messageContext.complete(); |
62 | | - } catch (Exception ex) { |
63 | | - LOGGER.logThrowableAsWarning(ex); |
| 161 | + |
| 162 | + LOGGER.atInfo() |
| 163 | + .addKeyValue("messageId", messageContext.getMessage().getMessageId()) |
| 164 | + .addKeyValue("deliveryCount", messageContext.getMessage().getDeliveryCount()) |
| 165 | + .log("message " + operation); |
| 166 | + } catch (RuntimeException ex) { |
| 167 | + runResult.set(RunResult.ERROR); |
| 168 | + LOGGER.atVerbose() |
| 169 | + .addKeyValue("messageId", messageContext.getMessage().getMessageId()) |
| 170 | + .log("message settlement failed"); |
64 | 171 | } |
65 | 172 | } |
66 | 173 | } |
0 commit comments