Add evaluator to alerting system - #6
Conversation
📝 WalkthroughWalkthroughAdds a Kafka-backed alert evaluator with YAML configuration, geofence rule detection, Prometheus metrics, lifecycle management, and integration tests. Updates the parallel consumer to use explicit topics and offset-based acknowledgements, with supporting client, dependency, protobuf, and test-infrastructure changes. ChangesEvaluator Runtime
Parallel Consumer Offset API
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant KafkaParallelConsumer
participant Evaluator
participant RuleViolationDetector
participant KafkaProducer
KafkaParallelConsumer->>Evaluator: poll protobuf Log
Evaluator->>RuleViolationDetector: findViolatedRules(Log)
RuleViolationDetector-->>Evaluator: return rule results
Evaluator->>KafkaProducer: produce TargetLog
KafkaProducer-->>Evaluator: acknowledge OffsetPartition on callback
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
b46163b to
83ea494
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (4)
alerting-system/client/src/main/java/ir/pathlens/alerting/client/RulesCache.java (1)
92-92: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winMake
RulesClientownership explicit.
RulesCachereceives an externally constructed client but now closes it, whileRuleControllerTest.javaalso closes the same instance at Line 82. Choose one owner—either haveRulesCachecreate/own the client or keep client shutdown at the composition root and document that injected clients are borrowed; otherwise shared callers can fail withIllegalStateExceptionafter cache shutdown.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@alerting-system/client/src/main/java/ir/pathlens/alerting/client/RulesCache.java` at line 92, Make ownership of the injected RulesClient explicit between RulesCache and the composition root: treat externally supplied clients as borrowed by removing client.close() from RulesCache, and document that callers retain shutdown responsibility. Ensure RuleControllerTest and other composition-root code remain responsible for closing the shared client.alerting-system/evaluator/build.gradle (1)
13-20: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winCentralize the evaluator dependency versions.
These dependencies pin versions inline while the repository already maintains
gradle/libs.versions.tomlas a version source. Add catalog aliases and consume them here to avoid version drift, especially between the existing Jackson catalog version and the inline Jackson 2.20.0 version. This is based on the suppliedgradle/libs.versions.tomldependency catalog.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@alerting-system/evaluator/build.gradle` around lines 13 - 20, Update the evaluator dependencies in build.gradle to use aliases from gradle/libs.versions.toml instead of inline versions, adding any missing catalog entries for Micrometer, Prometheus, SnakeYAML, and Jackson. Reuse the existing Jackson catalog version where applicable, ensuring both Jackson dependencies resolve through the catalog and no inline version pins remain.alerting-system/rest/src/main/java/ir/pathlens/alerting/rest/controller/MockRuleController.java (1)
1-82: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winTest-only mock server lives in the
restmodule'smainsourceSet.
MockRuleControlleris a test utility (per its Javadoc) but is placed undersrc/main/java, so it ships in therestmodule's production artifact. Since it's consumed cross-module byEvaluatorTest, consider using Gradle'sjava-test-fixturesplugin to expose it from a test-fixtures sourceSet instead, keeping it out of the production jar.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@alerting-system/rest/src/main/java/ir/pathlens/alerting/rest/controller/MockRuleController.java` around lines 1 - 82, Move MockRuleController from the rest module’s main source set into a test-fixtures source set, enable Gradle’s java-test-fixtures support, and update EvaluatorTest or other consumers to depend on the rest test-fixtures artifact. Keep the mock server available for cross-module tests while excluding it from the production jar.alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java (1)
100-101: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAvoid per-record
RuleViolationDetectorallocation on the hot path.
new RuleViolationDetector(rulesCache)is created for every consumed record even thoughrulesCachedoesn't change; consider hoisting it to a field created once in the constructor.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java` around lines 100 - 101, Hoist RuleViolationDetector creation out of processRecord and initialize a single instance in Evaluator’s constructor using rulesCache. Store it in a field and reuse that instance for every record, removing the per-record allocation while preserving existing detection behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java`:
- Around line 74-86: Update Evaluator.close so shared resources are not closed
immediately after executor.shutdownNow() when termination times out. Use the
awaitTermination result to guard or otherwise coordinate
parallelConsumer.close(), ensuring the poll loop has exited before closing the
consumer while preserving the existing shutdown sequence for successful
termination.
- Around line 100-162: Update processRecord, sendAlert, and getCallBack so each
input record is acknowledged exactly once: immediately ack offsetPartition when
both violatedRules() and nonViolatedRules() are empty, and otherwise send all
generated alerts before registering a single callback that acknowledges only
after every alert for that record succeeds. Handle retriable callback failures
with visible logging and preserve the existing fatal behavior for non-retriable
failures, including targetLog context such as ruleId and timestamp in error
messages.
In
`@common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java`:
- Around line 101-103: Move the drainCommittedMessages() call out of the
per-record/non-empty records path so it runs after every Kafka poll, including
empty polls, while preserving record tracking behavior. Add a regression test
for KafkaParallelConsumer that asynchronously acknowledges the final record,
allows at least one empty poll without closing the consumer, and verifies the
offset is committed.
In `@logs-proto/camera/proto/src/main/proto/log.proto`:
- Around line 34-35: Update the protobuf field migration around latitude and
longitude without reusing field numbers: assign new field numbers for the double
fields, preserve the original fields for compatibility, and implement
dual-read/dual-write behavior or coordinate a complete data migration before
changing the Log wire contract.
---
Nitpick comments:
In
`@alerting-system/client/src/main/java/ir/pathlens/alerting/client/RulesCache.java`:
- Line 92: Make ownership of the injected RulesClient explicit between
RulesCache and the composition root: treat externally supplied clients as
borrowed by removing client.close() from RulesCache, and document that callers
retain shutdown responsibility. Ensure RuleControllerTest and other
composition-root code remain responsible for closing the shared client.
In `@alerting-system/evaluator/build.gradle`:
- Around line 13-20: Update the evaluator dependencies in build.gradle to use
aliases from gradle/libs.versions.toml instead of inline versions, adding any
missing catalog entries for Micrometer, Prometheus, SnakeYAML, and Jackson.
Reuse the existing Jackson catalog version where applicable, ensuring both
Jackson dependencies resolve through the catalog and no inline version pins
remain.
In
`@alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java`:
- Around line 100-101: Hoist RuleViolationDetector creation out of processRecord
and initialize a single instance in Evaluator’s constructor using rulesCache.
Store it in a field and reuse that instance for every record, removing the
per-record allocation while preserving existing detection behavior.
In
`@alerting-system/rest/src/main/java/ir/pathlens/alerting/rest/controller/MockRuleController.java`:
- Around line 1-82: Move MockRuleController from the rest module’s main source
set into a test-fixtures source set, enable Gradle’s java-test-fixtures support,
and update EvaluatorTest or other consumers to depend on the rest test-fixtures
artifact. Keep the mock server available for cross-module tests while excluding
it from the production jar.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 1eb9d457-478f-48e3-9744-9ecda74e65c2
📒 Files selected for processing (26)
.idea/gradle.xmlalerting-system/client/src/main/java/ir/pathlens/alerting/client/RulesCache.javaalerting-system/client/src/main/java/ir/pathlens/alerting/client/RulesClient.javaalerting-system/evaluator/build.gradlealerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/ConfigReader.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/EvaluatorMain.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Profiler.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/RuleViolationDetector.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/configs/ApplicationConfig.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/configs/KafkaConsumerConfig.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/configs/KafkaProducerConfig.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/configs/RulesCacheConfig.javaalerting-system/evaluator/src/test/java/ir/pathlens/alerting/evaluator/EvaluatorTest.javaalerting-system/evaluator/src/test/resources/application.ymlalerting-system/rest/src/main/java/ir/pathlens/alerting/rest/controller/MockRuleController.javaalerting-system/rest/src/test/java/ir/pathlens/alerting/rest/controllers/RuleControllerTest.javacommon-libs/parallel-consumer/build.gradlecommon-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.javacommon-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/OffsetPartition.javacommon-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/Tracker.javacommon-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.javacommon-libs/test-extensions/build.gradlecommon-libs/test-extensions/src/main/java/ir/pathlens/extension/kafka/KafkaExtension.javagradle/libs.versions.tomllogs-proto/camera/proto/src/main/proto/log.proto
| @Override | ||
| public void close() throws Exception { | ||
| running = false; | ||
| executor.shutdown(); | ||
|
|
||
| if (!executor.awaitTermination(5, TimeUnit.SECONDS)) { | ||
| executor.shutdownNow(); | ||
| } | ||
|
|
||
| parallelConsumer.close(); | ||
| kafkaProducer.close(); | ||
| rulesCache.close(); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Consumer close proceeds even if the poll thread didn't confirm termination.
If awaitTermination times out and shutdownNow() is called, parallelConsumer.close() still runs immediately afterward without confirming pollLoop actually exited, risking a close-while-polling race in the (rare) hang case. Consider logging/guarding on the awaitTermination result before closing shared resources.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java`
around lines 74 - 86, Update Evaluator.close so shared resources are not closed
immediately after executor.shutdownNow() when termination times out. Use the
awaitTermination result to guard or otherwise coordinate
parallelConsumer.close(), ensuring the poll loop has exited before closing the
consumer while preserving the existing shutdown sequence for successful
termination.
| private void processRecord(Log log, OffsetPartition offsetPartition) { | ||
| RuleViolationDetector ruleViolationDetector = new RuleViolationDetector(rulesCache); | ||
| Result result = ruleViolationDetector.findViolatedRules(log); | ||
| if (!result.violatedRules().isEmpty() || !result.nonViolatedRules().isEmpty()) { | ||
| profiler.recordMatchedLog(); | ||
| } | ||
| profiler.recordViolatedRules(result.violatedRules().size()); | ||
| profiler.recordNonViolatedRules(result.nonViolatedRules().size()); | ||
| for (UUID ruleId : result.violatedRules()) { | ||
| TargetLog targetLog = TargetLog.newBuilder() | ||
| .setRuleId(ruleId.toString()) | ||
| .setTimestamp(log.getTimestamp()) | ||
| .setViolated(true) | ||
| .setLocation(Location.newBuilder() | ||
| .setLatitude(log.getLocation().getLatitude()) | ||
| .setLongitude(log.getLocation().getLongitude()) | ||
| .build()) | ||
| .build(); | ||
| sendAlert(targetLog, offsetPartition); | ||
| } | ||
| for (UUID ruleId : result.nonViolatedRules()) { | ||
| TargetLog targetLog = TargetLog.newBuilder() | ||
| .setRuleId(ruleId.toString()) | ||
| .setTimestamp(log.getTimestamp()) | ||
| .setViolated(false) | ||
| .setLocation(Location.newBuilder() | ||
| .setLatitude(log.getLocation().getLatitude()) | ||
| .setLongitude(log.getLocation().getLongitude()) | ||
| .build()) | ||
| .build(); | ||
| sendAlert(targetLog, offsetPartition); | ||
| } | ||
| } | ||
|
|
||
| private Log parseLog(byte[] record) { | ||
| try { | ||
| return Log.parseFrom(record); | ||
| } catch (InvalidProtocolBufferException e) { | ||
| throw new AssertionError("unexpected error happened, can not parse log", e); | ||
| } | ||
| } | ||
|
|
||
| private void sendAlert(TargetLog targetLog, OffsetPartition offsetPartition) { | ||
| ProducerRecord<byte[], byte[]> record = new ProducerRecord<>( | ||
| applicationConfig.destinationTopic(), targetLog.toByteArray()); | ||
| kafkaProducer.send(record, getCallBack(applicationConfig.destinationTopic(), targetLog, offsetPartition)); | ||
| } | ||
|
|
||
| private Callback getCallBack(String targetTopic, TargetLog targetLog, OffsetPartition offsetPartition) { | ||
| return (metadata, exception) -> { | ||
| if (exception == null) { | ||
| parallelConsumer.ack(offsetPartition); | ||
| return; | ||
| } | ||
| if (!(exception instanceof RetriableException)) { | ||
| logger.error( | ||
| "Crash because of an error on sending record to kafka targetTopicName: " + "{} .", targetTopic, | ||
| exception); | ||
| // Note that unlike the exit() method, halt() method does not cause shutdown hooks to be started. | ||
| Runtime.getRuntime().halt(-1); | ||
| } | ||
| }; | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
Offset acknowledgement logic is incomplete and can lose alerts / stall partition progress.
Two correctness problems in the ack flow between processRecord and sendAlert/getCallBack:
- Unmatched logs never get acked. If
result.violatedRules()andresult.nonViolatedRules()are both empty (the common case for logs whose identity has no configured rule), neither loop body runs, sosendAlert— and thereforeparallelConsumer.ack(offsetPartition)— is never called for that record. Since the parallel consumer commits based on acked offsets, this permanently stalls offset-commit progress on the partition: every "no-match" log becomes an un-acked gap, and on any restart/rebalance the consumer will reprocess everything from that gap onward. - Premature/partial ack for multi-rule records. For a record matching several rules, each
sendAlertcall independently acks the sameoffsetPartitionas soon as its own producer callback succeeds. If the process crashes after the first rule's alert is acked but before a second rule's alert for the same record is sent, that alert is permanently lost on restart (the offset is already past this record).
Additionally, in getCallBack: the exception instanceof RetriableException branch is a no-op — no log, no ack — so exhausted-retry failures are silently dropped with zero visibility, and the targetLog parameter is never used, losing useful context (ruleId/timestamp) for the error message that is logged.
🐛 Proposed fix: ack once per record only after all its alerts are sent, and always ack when there's nothing to send
private void processRecord(Log log, OffsetPartition offsetPartition) {
RuleViolationDetector ruleViolationDetector = new RuleViolationDetector(rulesCache);
Result result = ruleViolationDetector.findViolatedRules(log);
if (!result.violatedRules().isEmpty() || !result.nonViolatedRules().isEmpty()) {
profiler.recordMatchedLog();
}
profiler.recordViolatedRules(result.violatedRules().size());
profiler.recordNonViolatedRules(result.nonViolatedRules().size());
- for (UUID ruleId : result.violatedRules()) {
+ int totalAlerts = result.violatedRules().size() + result.nonViolatedRules().size();
+ if (totalAlerts == 0) {
+ parallelConsumer.ack(offsetPartition);
+ return;
+ }
+ AtomicInteger pendingSends = new AtomicInteger(totalAlerts);
+ for (UUID ruleId : result.violatedRules()) {
TargetLog targetLog = TargetLog.newBuilder()
.setRuleId(ruleId.toString())
.setTimestamp(log.getTimestamp())
.setViolated(true)
.setLocation(Location.newBuilder()
.setLatitude(log.getLocation().getLatitude())
.setLongitude(log.getLocation().getLongitude())
.build())
.build();
- sendAlert(targetLog, offsetPartition);
+ sendAlert(targetLog, offsetPartition, pendingSends);
}
- for (UUID ruleId : result.nonViolatedRules()) {
+ for (UUID ruleId : result.nonViolatedRules()) {
...
- sendAlert(targetLog, offsetPartition);
+ sendAlert(targetLog, offsetPartition, pendingSends);
}
}
- private void sendAlert(TargetLog targetLog, OffsetPartition offsetPartition) {
+ private void sendAlert(TargetLog targetLog, OffsetPartition offsetPartition, AtomicInteger pendingSends) {
ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(
applicationConfig.destinationTopic(), targetLog.toByteArray());
- kafkaProducer.send(record, getCallBack(applicationConfig.destinationTopic(), targetLog, offsetPartition));
+ kafkaProducer.send(record,
+ getCallBack(applicationConfig.destinationTopic(), targetLog, offsetPartition, pendingSends));
}
- private Callback getCallBack(String targetTopic, TargetLog targetLog, OffsetPartition offsetPartition) {
+ private Callback getCallBack(String targetTopic, TargetLog targetLog, OffsetPartition offsetPartition,
+ AtomicInteger pendingSends) {
return (metadata, exception) -> {
if (exception == null) {
- parallelConsumer.ack(offsetPartition);
+ if (pendingSends.decrementAndGet() == 0) {
+ parallelConsumer.ack(offsetPartition);
+ }
return;
}
if (!(exception instanceof RetriableException)) {
logger.error(
- "Crash because of an error on sending record to kafka targetTopicName: " + "{} .", targetTopic,
+ "Crash because of an error sending targetLog {} to kafka targetTopicName: {} .", targetLog,
exception);
Runtime.getRuntime().halt(-1);
+ } else {
+ logger.warn("Retriable error sending targetLog {} to {}, offset {} will not be acked.",
+ targetLog, targetTopic, offsetPartition, exception);
}
};
}🧰 Tools
🪛 PMD (7.26.0)
[Low] 155-157: InvalidLogMessageFormat (Error Prone): Too many arguments, expected 1 argument but found 2
(InvalidLogMessageFormat (Error Prone))
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java`
around lines 100 - 162, Update processRecord, sendAlert, and getCallBack so each
input record is acknowledged exactly once: immediately ack offsetPartition when
both violatedRules() and nonViolatedRules() are empty, and otherwise send all
generated alerts before registering a single callback that acknowledges only
after every alert for that record succeeds. Handle retriable callback failures
with visible logging and preserve the existing fatal behavior for non-retriable
failures, including targetLog context such as ruleId and timestamp in error
messages.
| double latitude = 1; | ||
| double longitude = 2; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Locate log.proto and Evaluator.java =="
fd -a 'log\.proto|Evaluator\.java' . | sed 's#^\./##'
echo
echo "== log.proto =="
file=$(fd 'log\.proto' . | head -n 1)
if [ -n "${file:-}" ]; then
nl -ba "$file" | sed -n '1,90p'
fi
echo
echo "== Evaluator.java likely matches =="
for f in $(fd 'Evaluator\.java' .); do
echo "--- $f ---"
nl -ba "$f" | sed -n '1,220p'
done
echo
echo "== Search coordinate usage =="
rg -n 'latitude|longitude|latitude|evaluate|coordinates|Coordinate|Log\(' -S .Repository: alishazaee/PathLens
Length of output: 568
🌐 Web query:
Protocol Buffers wire types float double fixed32 fixed64
💡 Result:
In Protocol Buffers, the wire type determines how a field is serialized on the wire [1][2]. The types float, double, fixed32, and fixed64 use fixed-width wire types, meaning they always consume a set number of bytes regardless of the value being stored [1][3][4]. The wire types are categorized as follows: Wire Type 5 (32-bit): Used for fixed32, sfixed32, and float [1][4]. These fields always occupy exactly 4 bytes of data in little-endian format [1][4]. - float: Encoded as an IEEE 754 single-precision floating-point number [1][3]. - fixed32: Encoded as a raw 32-bit unsigned integer [1][4]. Wire Type 1 (64-bit): Used for fixed64, sfixed64, and double [1][4]. These fields always occupy exactly 8 bytes of data in little-endian format [1][4]. - double: Encoded as an IEEE 754 double-precision floating-point number [1][3]. - fixed64: Encoded as a raw 64-bit unsigned integer [1][4]. Because these types use fixed-width encoding, they are often more efficient than variable-length (varint) encoding when the values are consistently large (e.g., fixed32 values frequently exceeding 2^28 or fixed64 values exceeding 2^56) [3][4][5]. Every field in the binary format is prefixed by a tag that combines the field number and the wire type into a single varint [2][4]. The wire type portion (the lower 3 bits) informs the decoder how many bytes to read for the subsequent value [2][4].
Citations:
- 1: https://protobuf.dev/programming-guides/encoding/
- 2: https://www.mintlify.com/protocolbuffers/protobuf/concepts/encoding
- 3: https://www.mintlify.com/protocolbuffers/protobuf/concepts/field-types
- 4: https://www.mintlify.com/protocolbuffers/protobuf/reference/encoding
- 5: https://protobuf.dev/programming-guides/proto3/
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== log.proto current =="
file="logs-proto/camera/proto/src/main/proto/log.proto"
if [ -f "$file" ]; then
awk '{printf "%6d\t%s\n", NR, $0}' "$file" | sed -n '1,90p'
fi
echo
echo "== Evaluator.java relevant coordinates =="
awk '{printf "%6d\t%s\n", NR, $0}' evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java | sed -n '1,260p'
echo
echo "== Search for double/float latitude/longitude in proto/java =="
rg -n 'double (latitude|longitude)|(latitude|longitude) = [0-9]|latitude\(\)|longitude\(\)|getLatitude|getLongitude' \
logs-proto/camera/proto/src/main/proto/log.proto evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java \
protobuf/src/main/proto/log.proto logs-proto/raw-log/proto/src/main/proto/log.proto 2>/dev/null || true
echo
echo "== Git diff stat/name for proto =="
git diff --stat -- logs-proto/camera/proto/src/main/proto/log.proto || true
git diff --unified=10 -- logs-proto/camera/proto/src/main/proto/log.proto || trueRepository: alishazaee/PathLens
Length of output: 2322
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Search coordinate-related usages =="
rg -n 'latitude|longitude|location\.getLongitude|location\.getLatitude|longitude\(\)|latitude\(\)|double latitude|float latitude|double longitude|float longitude' \
evaluation/src/main evaluation/src/main/java evaluator/src/main evaluator/src/main/java \
. -g '!build/**' -g '!target/**' 2>/dev/null \
| head -n 200 || true
echo
echo "== Candidate evaluator/java source files =="
find . -maxdepth 5 \( -path '*/evaluator/src/main/java/**' -o -path '*/evaluation/src/main/java/**' \) -name '*.java' \
-type f 2>/dev/null | sort | sed -n '1,80p'
echo
echo "== Locate CameraLogProto =="
rg -n 'CameraLogProto|pathlens\.proto|pathlens\.camera' . -g '*.java' -g '*.scala' 2>/dev/null | head -n 80 || trueRepository: alishazaee/PathLens
Length of output: 15484
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Evaluator.java coordinate propagation =="
awk '{printf "%6d\t%s\n", NR, $0}' evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java | sed -n '1,220p'
echo
echo "== Detector coordinates =="
awk '{printf "%6d\t%s\n", NR, $0}' evaluator/src/main/java/ir/pathlens/alerting/evaluator/RuleViolationDetector.java | sed -n '1,140p'
echo
echo "== Camera parser location building =="
awk '{printf "%6d\t%s\n", NR, $0}' logs-proto/camera/parser/src/main/java/ir/pathlens/camera/CameraLogParser.java | sed -n '1,180p'
echo
echo "== Camera log generator location fields =="
awk '{printf "%6d\t%s\n", NR, $0}' common-libs/log-generator/src/main/java/ir/pathlens/generator/CameraLogGenerator.java | sed -n '1,160p'
echo
echo "== Proto comparator tests =="
if [ -f common-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.java ]; then
awk '{printf "%6d\t%s\n", NR, $0}' common-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.java | sed -n '1,120p'
fiRepository: alishazaee/PathLens
Length of output: 341
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Exact file exists =="
ls -l logs-proto/camera/proto/src/main/proto/log.proto
ls -l alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java
ls -l alerting-system/rest/src/main/java/ir/pathlens/alerting/rest/entity/LogEntity.java
echo
echo "== Evaluator.java coordinate propagation =="
awk '{printf "%6d\t%s\n", NR, $0}' alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java | sed -n '1,240p'
echo
echo "== Detector coordinates =="
awk '{printf "%6d\t%s\n", NR, $0}' alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/RuleViolationDetector.java | sed -n '1,150p'
echo
echo "== Camera parser location/building =="
awk '{printf "%6d\t%s\n", NR, $0}' logs-proto/camera/parser/src/main/java/ir/pathlens/camera/CameraLogParser.java | sed -n '1,190p'
echo
echo "== Camera log generator location fields =="
awk '{printf "%6d\t%s\n", NR, $0}' common-libs/log-generator/src/main/java/ir/pathlens/generator/CameraLogGenerator.java | sed -n '1,160p'
echo
echo "== Proto comparator/tests =="
awk '{printf "%6d\t%s\n", NR, $0}' common-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.java | sed -n '1,130p'Repository: alishazaee/PathLens
Length of output: 34135
Do not reuse field numbers when changing protobuf field wires.
float is wire type 5 (fixed32) and double is wire type 1 (fixed64), so keeping latitude = 1 and longitude = 2 while changing the wires makes existing Log payloads incompatible. Add new field numbers with dual-read/dual-write migration, or coordinate a complete data migration before changing this contract.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@logs-proto/camera/proto/src/main/proto/log.proto` around lines 34 - 35,
Update the protobuf field migration around latitude and longitude without
reusing field numbers: assign new field numbers for the double fields, preserve
the original fields for compatibility, and implement dual-read/dual-write
behavior or coordinate a complete data migration before changing the Log wire
contract.
83ea494 to
c5e4ebf
Compare
c5e4ebf to
52138df
Compare
There was a problem hiding this comment.
Actionable comments posted: 8
🧹 Nitpick comments (3)
alerting-system/evaluator/build.gradle (1)
19-20: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse the version catalog for Jackson dependencies.
gradle/libs.versions.tomlalready defines Jackson aliases, but these declarations hard-code2.20.0while the catalog specifies2.15.2. Update the catalog if 2.20.0 is intended, then uselibs.jackson.databindandlibs.jackson.yamlhere to keep versions aligned.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@alerting-system/evaluator/build.gradle` around lines 19 - 20, Replace the hard-coded Jackson dependencies in the evaluator build configuration with the version-catalog aliases libs.jackson.databind and libs.jackson.yaml. If Jackson 2.20.0 is required, update the catalog’s Jackson version first; otherwise preserve its existing 2.15.2 version so both declarations remain centrally aligned.alerting-system/evaluator/src/test/java/ir/pathlens/alerting/evaluator/EvaluatorTest.java (2)
229-232: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win
produceLogignores send failures.
producer.send(producerRecord)discards the returnedFutureand any exception. If a send fails, the test won't surface the actual cause — it will just time out in theAwaitilityblock waiting for records that never arrive. Consider using a callback or.get()to fail fast with the real error.🔧 Proposed fix
private void produceLog(String topic, byte[] value) { ProducerRecord<byte[], byte[]> producerRecord = new ProducerRecord<>(topic, value); - producer.send(producerRecord); + producer.send(producerRecord, (metadata, exception) -> { + if (exception != null) { + throw new RuntimeException("Failed to produce test log", exception); + } + }); }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@alerting-system/evaluator/src/test/java/ir/pathlens/alerting/evaluator/EvaluatorTest.java` around lines 229 - 232, Update produceLog to handle the Future returned by producer.send(producerRecord), waiting for completion or registering a failure callback so send exceptions surface immediately with their original cause instead of only causing an Awaitility timeout.
105-118: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicate polling/collection logic across both tests.
The
Awaitility.await()...untilAssertedblock that pollskafkaConsumer, parsesTargetLogProto.TargetLog, and accumulatesoutputTargetsis duplicated verbatim in both tests. Extract a private helper (e.g.,collectOutputTargets(int expectedCount)) to avoid drift between the two copies.♻️ Proposed helper extraction
+ private List<TargetLogProto.TargetLog> collectOutputTargets(int expectedCount) { + List<TargetLogProto.TargetLog> outputTargets = new ArrayList<>(); + Awaitility.await() + .pollInterval(Duration.ofMillis(50)) + .atMost(Duration.ofSeconds(5)) + .untilAsserted(() -> { + ConsumerRecords<byte[], byte[]> records = kafkaConsumer.poll(Duration.ofMillis(100)); + for (ConsumerRecord<byte[], byte[]> record : records) { + if (record.value() != null) { + outputTargets.add(TargetLogProto.TargetLog.parseFrom(record.value())); + } + } + assertEquals(expectedCount, outputTargets.size()); + }); + return outputTargets; + }Also applies to: 161-174
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@alerting-system/evaluator/src/test/java/ir/pathlens/alerting/evaluator/EvaluatorTest.java` around lines 105 - 118, Extract the duplicated Awaitility polling and TargetLogProto.TargetLog collection into a private helper such as collectOutputTargets(int expectedCount). Have the helper poll kafkaConsumer, parse non-null records, accumulate targets, and assert the expected count, then update both tests to call it and use the returned collection.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java`:
- Line 73: Update the pollLoop boundary around parseLog so malformed Protobuf
input cannot terminate polling silently. Catch decode failures within the
polling flow, send the malformed message to the established DLQ and acknowledge
it only after that send succeeds, or deliberately propagate the failure through
the evaluator’s lifecycle instead of relying on the ignored Future from
executor.submit.
- Around line 183-186: Update the consumer property setup in Evaluator so
kafkaConsumerConfig.getExtraConfigs() is applied before setting
ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to false. Preserve evaluator ownership
of manual commit mode by enforcing the false value after all extra
configurations are merged.
In
`@alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/RuleViolationDetector.java`:
- Around line 77-82: Update the geometry parsing flow in processRecord to handle
WKT ParseException as a runtime/configuration error instead of wrapping it in
AssertionError. Isolate the malformed rule using its rule ID, such as by logging
or skipping that rule, while preserving evaluation for other rules;
alternatively validate the WKT during cache refresh.
In
`@alerting-system/evaluator/src/test/java/ir/pathlens/alerting/evaluator/EvaluatorTest.java`:
- Line 44: The tests currently import Testcontainers’ internal shaded Awaitility
API. Add an explicit org.awaitility:awaitility test dependency, replace the
import in EvaluatorTest with org.awaitility.Awaitility, and update all
Awaitility call sites in the test to use that public dependency.
In
`@alerting-system/rest/src/main/java/ir/pathlens/alerting/rest/controller/MockRuleController.java`:
- Around line 23-81: Move the test-only MockRuleController class out of the rest
module’s production sources into the shared test-extensions module or the rest
module’s test-fixtures source set. Update evaluator test setup and any imports
or dependencies to use the relocated class, preserving its REST endpoints and
ephemeral-port base URL behavior.
- Around line 3-4: Update the Jackson dependencies in
alerting-system/rest/build.gradle and build.gradle.kts to use the
version-catalog Jackson coordinates rather than hardcoded versions, and align
ObjectMapper/JavaTimeModule with the catalog-managed version. Ensure the shared
Jackson version is upgraded beyond the vulnerable 2.15.x/2.20.x lines across the
affected modules.
In
`@common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java`:
- Around line 104-107: Update the record-processing loop in
KafkaParallelConsumer.pollLoop so observing running == false while retrying
polledRecords.offer(record) exits pollLoop immediately rather than only breaking
the inner retry loop; return from the method so its existing finally block still
executes.
In
`@common-libs/test-extensions/src/main/java/ir/pathlens/extension/kafka/KafkaExtension.java`:
- Around line 38-53: Update KafkaExtension’s beforeEach/afterEach topic-cleanup
design so concurrent tests sharing the JVM KafkaContainer cannot delete topics
created by other executions; replace the per-test snapshot deletion in
beforeEach, afterEach, and deleteNewTopics with execution isolation or
synchronization that protects each test class/context’s topics. Preserve cleanup
of topics created by the current execution while preventing cross-test deletion
during parallel runs.
---
Nitpick comments:
In `@alerting-system/evaluator/build.gradle`:
- Around line 19-20: Replace the hard-coded Jackson dependencies in the
evaluator build configuration with the version-catalog aliases
libs.jackson.databind and libs.jackson.yaml. If Jackson 2.20.0 is required,
update the catalog’s Jackson version first; otherwise preserve its existing
2.15.2 version so both declarations remain centrally aligned.
In
`@alerting-system/evaluator/src/test/java/ir/pathlens/alerting/evaluator/EvaluatorTest.java`:
- Around line 229-232: Update produceLog to handle the Future returned by
producer.send(producerRecord), waiting for completion or registering a failure
callback so send exceptions surface immediately with their original cause
instead of only causing an Awaitility timeout.
- Around line 105-118: Extract the duplicated Awaitility polling and
TargetLogProto.TargetLog collection into a private helper such as
collectOutputTargets(int expectedCount). Have the helper poll kafkaConsumer,
parse non-null records, accumulate targets, and assert the expected count, then
update both tests to call it and use the returned collection.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: f0e2a582-85f5-4fcd-be0b-967387d173b3
📒 Files selected for processing (27)
.idea/gradle.xmlalerting-system/client/src/main/java/ir/pathlens/alerting/client/RulesCache.javaalerting-system/client/src/main/java/ir/pathlens/alerting/client/RulesClient.javaalerting-system/evaluator/build.gradlealerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/ConfigReader.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/EvaluatorMain.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Profiler.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/RuleViolationDetector.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/configs/ApplicationConfig.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/configs/KafkaConsumerConfig.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/configs/KafkaProducerConfig.javaalerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/configs/RulesCacheConfig.javaalerting-system/evaluator/src/test/java/ir/pathlens/alerting/evaluator/EvaluatorTest.javaalerting-system/evaluator/src/test/resources/application.ymlalerting-system/rest/src/main/java/ir/pathlens/alerting/rest/controller/MockRuleController.javaalerting-system/rest/src/test/java/ir/pathlens/alerting/rest/controllers/RuleControllerTest.javacommon-libs/parallel-consumer/build.gradlecommon-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.javacommon-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/OffsetPartition.javacommon-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/Tracker.javacommon-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.javacommon-libs/test-extensions/build.gradlecommon-libs/test-extensions/src/main/java/ir/pathlens/extension/kafka/KafkaExtension.javagradle/libs.versions.tomllogs-proto/camera/parser/src/test/java/ir/pathlens/camera/TestCameraLogParser.javalogs-proto/camera/proto/src/main/proto/log.proto
🚧 Files skipped from review as they are similar to previous changes (3)
- alerting-system/client/src/main/java/ir/pathlens/alerting/client/RulesClient.java
- common-libs/parallel-consumer/build.gradle
- .idea/gradle.xml
| rulesCache.submitBackgroundTask(); | ||
| rulesCache.sync(); | ||
| parallelConsumer.start(applicationConfig.sourceTopic()); | ||
| executor.submit(this::pollLoop); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== git status/stat =="
git diff --stat || true
echo "== locate Evaluator.java =="
fd -a 'Evaluator\.java$' . | sed 's#^\./##'
echo "== outline Evaluator =="
ast-grep outline alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java --view expanded || true
echo "== relevant source sections =="
sed -n '1,190p' alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java | cat -nRepository: alishazaee/PathLens
Length of output: 11624
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Java Executor submit/Future behavior probe =="
if command -v javac >/dev/null 2>&1 && command -v java >/dev/null 2>&1; then
cat >/tmp/SubmitBehavior.java <<'JAVA'
import java.util.concurrent.*;
public class SubmitBehavior {
public static void main(String[] args) throws InterruptedException {
Thread.setDefaultUncaughtExceptionHandler((t, e) -> System.out.println("UNCAUGHT: " + e.getClass().getName()));
ExecutorService executor = Executors.newFixedThreadPool(1);
try {
Callable<Void> task = () -> { throw new Error("probe"); };
Future<?> f = executor.submit(task);
try {
f.get(250, TimeUnit.MILLISECONDS);
System.out.println("GET_RETURNED_NORMAL");
} catch (ExecutionException ex) {
System.out.println("GET_THREW_EXECUTION");
System.out.println(getCauseClass(ex));
}
System.out.println("ALIVE_AFTER_SUBMIT_EXCEPTION=" + executor.isTerminated());
} finally {
executor.shutdownNow();
}
Thread.sleep(500);
}
static String getCauseClass(ExecutionException e) {
Throwable c = e.getCause();
return c != null ? c.getClass().getName() : "NO_CAUSE";
}
}
JAVA
javac /tmp/SubmitBehavior.java
java -cp /tmp SubmitBehavior
else
echo "javac/java unavailable"
fi
echo "== parseLog/probability of parseFrom throwing for malformed bytes =="
python3 - <<'PY'
import os
paths = ["alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java"]
for p in paths:
text = open(p, encoding="utf-8").read()
i = text.find("Log.parseFrom(record)")
print(f"parseFrom_at_line={text[:i].count(chr(10))+1 if i!=-1 else -1}")
print(text[i:i+300])
PYRepository: alishazaee/PathLens
Length of output: 621
🌐 Web query:
Java 8 ExecutorService submit Callable returns Future exception execution does not invoke uncaught exception handler
💡 Result:
When you submit a task to an ExecutorService using the submit method, any exception thrown by that task is captured and stored by the returned Future object, rather than being treated as an uncaught exception [1][2]. Because the framework explicitly handles the exception as part of the task's result, it does not propagate to the thread's UncaughtExceptionHandler [1][3][4]. If you wish to access or handle the exception, you must do so via the Future object: 1. When you call future.get, the captured exception is rethrown wrapped in an ExecutionException [5][6][2]. 2. You can retrieve the original exception by calling getCause on that ExecutionException [2][7]. In contrast, the execute method (which returns void and accepts only Runnable tasks) does not return a Future [8][2]. Consequently, it has no way to return an exception to the caller, so any uncaught exception is passed to the thread's UncaughtExceptionHandler [1][2]. If you require global or centralized exception handling for tasks submitted via submit, you can subclass ThreadPoolExecutor and override the afterExecute(Runnable r, Throwable t) method [1][9][7]. In this override, if t is null, you can check if r is a Future and attempt to retrieve the result (and potential exception) using Future.get to perform your own logging or error handling [9][7].
Citations:
- 1: https://stackoverflow.com/questions/1838923/why-is-uncaughtexceptionhandler-not-called-by-executorservice
- 2: https://stackoverflow.com/questions/3929342/choose-between-executorservices-submit-and-executorservices-execute
- 3: https://www.javabyexamples.com/handling-exceptions-from-executorservice-tasks
- 4: https://bugs.openjdk.org/browse/JDK-8240148
- 5: https://community.sonarsource.com/t/java-always-handle-future-returned-by-executorservice-submit/20186
- 6: https://codemia.io/knowledge-hub/path/handling_exceptions_from_java_executorservice_tasks
- 7: https://stackoverflow.com/questions/1687977/how-to-properly-catch-runtimeexceptions-from-executors
- 8: https://docs.oracle.com/en/java/javase/26/docs/api/java.base/java/util/concurrent/ExecutorService.html
- 9: https://stackoverflow.com/questions/2248131/handling-exceptions-from-java-executorservice-tasks
🌐 Web query:
Java 8 Thread UncaughtExceptionHandler Executors.newFixedThreadPool rejectExecution rejected execution handler Executor
💡 Result:
In Java 8, handling uncaught exceptions and rejected tasks in an ExecutorService requires different strategies because they address distinct failure scenarios [1][2]. Uncaught Exceptions When using Executors.newFixedThreadPool, tasks submitted via execute that throw uncaught exceptions do not automatically trigger the Thread.UncaughtExceptionHandler [1]. This is because the Executor framework manages the lifecycle of threads and captures exceptions within the task itself [1][3]. To handle these exceptions, you have three primary options: 1. Use submit instead of execute: When you use submit, the task is wrapped in a FutureTask which captures the exception [3]. You can then call Future.get, which will throw an ExecutionException containing the original exception as its cause [3]. 2. Override afterExecute: You can extend ThreadPoolExecutor and override the protected void afterExecute(Runnable r, Throwable t) method [1][2]. This method is invoked by the thread that executed the task, and if the task ended abruptly, the Throwable argument will contain the exception [3]. 3. Custom ThreadFactory: While you can set an UncaughtExceptionHandler on threads created by a custom ThreadFactory, this is often insufficient for tasks submitted to an Executor because the Executor framework typically handles exceptions internally before they reach the thread level [4][1]. Rejected Execution A RejectedExecutionHandler is invoked when a task cannot be accepted by the executor, such as when the thread pool is saturated or has been shut down [2][5]. This is unrelated to exceptions thrown during the execution of a task [2]. To define how rejected tasks are handled, use the setRejectedExecutionHandler method on your ThreadPoolExecutor instance [2][6]. Java provides four built-in policies [2][5]: - AbortPolicy (default): Throws a RejectedExecutionException [2][5]. - CallerRunsPolicy: The thread that submitted the task executes it itself [2]. - DiscardPolicy: Silently drops the task [2]. - DiscardOldestPolicy: Drops the oldest unhandled task in the queue and retries submission [2]. You can also implement the RejectedExecutionHandler interface to create custom logic for handling rejected tasks (e.g., logging, saving to a persistent store, or blocking until space is available) [7][6].
Citations:
- 1: https://stackoverflow.com/questions/1838923/why-is-uncaughtexceptionhandler-not-called-by-executorservice
- 2: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ThreadPoolExecutor.html
- 3: https://stackoverflow.com/questions/1687977/how-to-properly-catch-runtimeexceptions-from-executors
- 4: https://stackoverflow.com/questions/24105323/catch-exception-for-executor-thread-pool
- 5: https://github.com/openjdk/jdk/blob/jdk8-b21/jdk/src/share/classes/java/util/concurrent/ThreadPoolExecutor.java
- 6: https://howtodoinjava.com/java/multi-threading/executor-rejectedexecutionhandler/
- 7: https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/RejectedExecutionHandler.html
Do not silently kill the poll loop on malformed input.
parseLog converts malformed Protobuf bytes into an unchecked AssertionError, and because executor.submit(this::pollLoop) ignores the returned Future, that exception is not surfaced from the submit call. The evaluator can stay up while pollLoop() never runs again; handle decode failures at the poll boundary instead, e.g. send to a DLQ and ack only after the DLQ send succeeds, or fail deliberately.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java`
at line 73, Update the pollLoop boundary around parseLog so malformed Protobuf
input cannot terminate polling silently. Catch decode failures within the
polling flow, send the malformed message to the established DLQ and acknowledge
it only after that send succeeds, or deliberately propagate the failure through
the evaluator’s lifecycle instead of relying on the ignored Future from
executor.submit.
| props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); | ||
| if (kafkaConsumerConfig.getExtraConfigs() != null) { | ||
| props.putAll(kafkaConsumerConfig.getExtraConfigs()); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== locate Evaluator and config classes =="
fd -a 'Evaluator\.java|KafkaConsumerConfig\.java|.*Consumer.*\.java' . | sed 's#^\./##'
echo "== relevant Evaluator excerpts =="
if [ -f alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java ]; then
wc -l alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java
sed -n '150,220p' alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java
fi
echo "== config class definitions/fields =="
for f in $(fd 'KafkaConsumerConfig\.java|ConsumerConfig\.java' .); do
echo "--- $f"
wc -l "$f"
sed -n '1,220p' "$f"
done
echo "== ack/commit-related usages =="
rg -n "parallelConsumer|ack|commit|ENABLE_AUTO_COMMIT_CONFIG|extraConfigs|getExtraConfigs" alerting-system/evaluator/src/main/java -SRepository: alishazaee/PathLens
Length of output: 12938
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== KafkaParallelConsumer outline =="
ast-grep outline alerting-system/common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java --view expanded || true
echo "== KafkaParallelConsumer relevant code =="
sed -n '1,260p' alerting-system/common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java
echo "== ack invocations and offset/commit state =="
rg -n "ack\\(|offsetPartition|commit|AutoCommit|ConsumerInterceptor|KafkaClient" alerting-system/evaluator/src/main/java alerting-system/common-libs/parallel-consumer/src/main/java -S
echo "== Java Properties.putAll precedence probe =="
cat > /tmp/PropertiesPutAllPrecedence.java <<'JAVA'
import java.util.Properties;
public class PropertiesPutAllPrecedence {
public static void main(String[] args) {
Properties props = new Properties();
props.put("enable.auto.commit", "false");
Properties extras = new Properties();
extras.put("enable.auto.commit", "true");
props.putAll(extras);
System.out.println(props.get("enable.auto.commit"));
}
}
JAVA
javac /tmp/PropertiesPutAllPrecedence.java
java -cp /tmp PropertiesPutAllPrecedenceRepository: alishazaee/PathLens
Length of output: 571
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== current repo files matching consumer class names =="
git ls-files | rg '(^|/)KafkaParallelConsumer\.java$|KafkaConsumerConfig\.java$|Evaluator\.java$|offsetPartition|Ack|ack\('
echo "== locate ack offsetPartition commit exact usages =="
rg -n "parallelConsumer\.ack|ack\\(|offsetPartition|commit|AutoCommit|ConsumerInterceptor|KafkaClient|ENABLE_AUTO_COMMIT_CONFIG" -S .Repository: alishazaee/PathLens
Length of output: 7725
Prevent extraConfigs from overriding manual commit mode.
extraConfigs.enable.auto.commit=true is applied after ENABLE_AUTO_COMMIT_CONFIG is set to false, which can make Kafka auto-commit offsets before parallelConsumer.ack(...) completes processing/target delivery. Apply extras first, then enforce evaluator-owned consumer properties.
Proposed fix
private static KafkaParallelConsumer<byte[], byte[]> createKafkaConsumer(KafkaConsumerConfig kafkaConsumerConfig) {
Properties props = new Properties();
+ if (kafkaConsumerConfig.getExtraConfigs() != null) {
+ props.putAll(kafkaConsumerConfig.getExtraConfigs());
+ }
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConsumerConfig.getBootstrapServers());
props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerConfig.getGroupId());
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaConsumerConfig.getAutoOffsetReset());
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
- if (kafkaConsumerConfig.getExtraConfigs() != null) {
- props.putAll(kafkaConsumerConfig.getExtraConfigs());
- }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); | |
| if (kafkaConsumerConfig.getExtraConfigs() != null) { | |
| props.putAll(kafkaConsumerConfig.getExtraConfigs()); | |
| } | |
| private static KafkaParallelConsumer<byte[], byte[]> createKafkaConsumer(KafkaConsumerConfig kafkaConsumerConfig) { | |
| Properties props = new Properties(); | |
| if (kafkaConsumerConfig.getExtraConfigs() != null) { | |
| props.putAll(kafkaConsumerConfig.getExtraConfigs()); | |
| } | |
| props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConsumerConfig.getBootstrapServers()); | |
| props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerConfig.getGroupId()); | |
| props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName()); | |
| props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName()); | |
| props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaConsumerConfig.getAutoOffsetReset()); | |
| props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java`
around lines 183 - 186, Update the consumer property setup in Evaluator so
kafkaConsumerConfig.getExtraConfigs() is applied before setting
ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to false. Preserve evaluator ownership
of manual commit mode by enforcing the false value after all extra
configurations are merged.
| Geometry geometry; | ||
| try { | ||
| geometry = wktReader.read(wktGeometry.get()); | ||
| } catch (ParseException e) { | ||
| throw new AssertionError("Unexpected error, parse failed", e); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate and inspect the target file and relevant nearby/call-site code.
if [ -f alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/RuleViolationDetector.java ]; then
echo "== RuleViolationDetector outline =="
ast-grep outline alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/RuleViolationDetector.java || true
echo "== RuleViolationDetector lines 1-190 =="
cat -n alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/RuleViolationDetector.java | sed -n '1,190p'
else
echo "target file not found"
fd -a 'RuleViolationDetector.java' . || true
fi
echo "== Search for RulesCache/wkt parsing/evaluation paths =="
rg -n "RulesCache|wktGeometry|WKT|ParseException|assertion|assert|RuleViolationDetector|alert|evaluate" alerting-system/evaluator/src/main/java -S || true
echo "== Diff stat and relevant diff snippets =="
git diff --stat || true
git diff -- alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/RuleViolationDetector.java | sed -n '1,260p' || trueRepository: alishazaee/PathLens
Length of output: 12780
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Read all files that directly use RuleViolationDetector or RulesCache geometry fields.
echo "== Candidate files using RuleViolationDetector or wktGeometry/RulesCache =="
rg -l "RuleViolationDetector|RulesCache|wktGeometry|Geometry" alerting-system/evaluator/src/main/java || true
echo "== Inspect files using RuleViolationDetector =="
while IFS= read -r f; do
echo "--- $f ---"
rg -n "RuleViolationDetector|create|detect|geometry|RulesCache" "$f" -A 5 -B 5 || true
done < <(rg -l "RuleViolationDetector" alerting-system/evaluator/src/main/java || true)
echo "== Inspect suspected RulesCache geometry fields =="
while IFS= read -r f; do
echo "--- $f ---"
ast-grep outline "$f" --view expanded || true
cat -n "$f" | sed -n '1,260p'
done < <(rg -l "class RulesCache|interface RulesCache|record RulesCache|RulesCache" alerting-system/evaluator/src/main/java || true)Repository: alishazaee/PathLens
Length of output: 721
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Evaluator outline and lines 1-180 =="
ast-grep outline alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java --view expanded || true
cat -n alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java | sed -n '1,180p'
echo "== RulesCache source files =="
fd -a 'RulesCache.java' alerting-system/evaluator/src/main/java || true
while IFS= read -r f; do
echo "--- $f ---"
ast-grep outline "$f" --view expanded || true
cat -n "$f" | sed -n '1,320p'
done < <(fd 'RulesCache.java' alerting-system/evaluator/src/main/java)
echo "== Application/config refs around RulesCache =="
cat -n alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/configs/ApplicationConfig.java | sed -n '1,220p'
cat -n alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/configs/RulesCacheConfig.java | sed -n '1,220p'Repository: alishazaee/PathLens
Length of output: 10827
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== RulesCache.java files =="
fd -a 'RulesCache.java' alerting-system/evaluator/src/main/java || true
while IFS= read -r f; do
echo "--- $f ---"
ast-grep outline "$f" --view expanded || true
cat -n "$f" | sed -n '1,360p'
done < <(fd -a 'RulesCache.java' alerting-system/evaluator/src/main/java || true)
echo "== Java assertion error behavior probe (if javac/java available) =="
if command -v java >/dev/null 2>&1 && command -v javac >/dev/null 2>&1; then
tmpdir="$(mktemp -d)"
cat > "$tmpdir/Probe.java" <<'JAVA'
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class Probe {
public static void main(String[] args) throws Exception {
Runnable task = () -> { throw new AssertionError("parse failed"); };
ExecutorService executor = Executors.newSingleThreadExecutor();
executor.submit(task);
executor.shutdownNow();
System.out.println(executor.awaitTermination(1, TimeUnit.SECONDS));
}
}
JAVA
javac "$tmpdir/Probe.java"
java -cp "$tmpdir" Probe
else
echo "javac/java not available"
fi
echo "== Behavioral/source verifier =="
python3 - <<'PY'
from pathlib import Path
target = Path('alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/RuleViolationDetector.java')
text = target.read_text()
checks = {
'catches ParseException': 'catch (ParseException e)' in text,
'throws AssertionError': 'throw new AssertionError("Unexpected error, parse failed", e)' in text,
'called by processRecord': Path('alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/Evaluator.java').read_text().find('ruleViolationDetector.findViolatedRules(log)') != -1,
}
for k, v in checks.items():
print(f'{k}: {v}')
PYRepository: alishazaee/PathLens
Length of output: 238
Do not convert malformed rule data into AssertionError.
The WKT comes from cached rules and is parsed during record evaluation, so parsing failures are runtime/configuration errors. Catch malformed rules in processRecord (or pre-validate during cache refresh) and isolate the failing rule ID rather than throwing an uncaught AssertionError in the evaluator path.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@alerting-system/evaluator/src/main/java/ir/pathlens/alerting/evaluator/RuleViolationDetector.java`
around lines 77 - 82, Update the geometry parsing flow in processRecord to
handle WKT ParseException as a runtime/configuration error instead of wrapping
it in AssertionError. Isolate the malformed rule using its rule ID, such as by
logging or skipping that rule, while preserving evaluation for other rules;
alternatively validate the WKT during cache refresh.
| import org.junit.jupiter.api.BeforeEach; | ||
| import org.junit.jupiter.api.Test; | ||
| import org.junit.jupiter.api.extension.ExtendWith; | ||
| import org.testcontainers.shaded.org.awaitility.Awaitility; |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
org.testcontainers shaded awaitility package stability across testcontainers versions
💡 Result:
Testcontainers shades Awaitility (relocating it to the org.testcontainers.shaded.org.awaitility package) as an internal implementation detail, not a public API [1][2]. Key stability and usage points include: 1. Not for Public Use: It is strongly recommended that you do not depend on the shaded version of Awaitility in your own test code [2]. Because these classes are relocated and bundled as a private implementation detail, they are subject to change, removal, or version updates at the maintainers' discretion without warning [2]. You should add an explicit, non-shaded dependency on the Awaitility library to your project's build configuration (e.g., pom.xml or build.gradle) if you need to use it [2]. 2. Shading Purpose: Testcontainers uses shading for dependencies like Awaitility, Guava, and others to prevent classpath conflicts with the application under test [1]. This ensures that Testcontainers' internal operations (such as waiting for container readiness) do not collide with different versions of the same libraries used in your project [1]. 3. Version Stability: While Testcontainers periodically bumps the version of the internal Awaitility dependency [3][4], these updates are managed internally. Using the shaded package directly ties your code to the specific, internal version bundled with the Testcontainers JAR, which makes your tests fragile and dependent on Testcontainers' internal dependency management [2]. 4. Potential for Conflicts: Although shading is designed to isolate dependencies, relying on shaded packages can occasionally lead to unexpected behavior or issues if there are conflicts with global JVM states or classloader assumptions [5][6][7]. For instance, Testcontainers has historically had to adjust how it uses Awaitility to avoid unintended side effects, such as temporarily overriding the global uncaught exception handler [7]. In summary, treat the shaded Awaitility package as invisible to your code. Always include your own dependency for Awaitility to ensure long-term stability and compatibility for your test suite [2].
Citations:
- 1: https://github.com/testcontainers/testcontainers-java/blob/main/docs/index.md
- 2: https://stackoverflow.com/questions/76874119/testcontainers-shaded-dependencies
- 3: Combined dependencies PR testcontainers/testcontainers-java#9284
- 4: Combined dependencies PR testcontainers/testcontainers-java#10457
- 5: [Bug]: Exception when running testcontainer from a groovy-maven-plugin script testcontainers/testcontainers-java#7923
- 6: [Bug]: Running testcontainers temporarily changes global uncaught exception handler testcontainers/testcontainers-java#11483
- 7: Don't replace the global uncaught exception handler during container startup testcontainers/testcontainers-java#11920
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Find build/test files:"
git ls-files | rg '(^|/)(build\.gradle(\.kts)?|pom\.xml|settings\.gradle(\.kts)?|gradle\.properties)$|EvaluatorTest\.java$' || true
echo
echo "References to Awaitility:"
rg -n "Awaitility|awaitility|testcontainers" -S . || trueRepository: alishazaee/PathLens
Length of output: 4517
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Version catalog:"
cat -n gradle/libs.versions.toml | sed -n '1,80p'
echo
echo "Evaluator build.gradle:"
cat -n alerting-system/evaluator/build.gradle | sed -n '1,180p'
echo
echo "Evaluators test imports/await calls:"
cat -n alerting-system/evaluator/src/test/java/ir/pathlens/alerting/evaluator/EvaluatorTest.java | sed -n '1,220p'Repository: alishazaee/PathLens
Length of output: 16440
Avoid relying on testcontainers' shaded Awaitility package.
org.testcontainers.shaded.org.awaitility.Awaitility is an internal repackaged dependency, so this import is fragile across Testcontainers updates. Add an explicit org.awaitility:awaitility test dependency and import org.awaitility.Awaitility instead. Also applies to the Awaitility call sites at lines 106-118 and 162-174.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@alerting-system/evaluator/src/test/java/ir/pathlens/alerting/evaluator/EvaluatorTest.java`
at line 44, The tests currently import Testcontainers’ internal shaded
Awaitility API. Add an explicit org.awaitility:awaitility test dependency,
replace the import in EvaluatorTest with org.awaitility.Awaitility, and update
all Awaitility call sites in the test to use that public dependency.
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
jackson-databind 2.20.0 known CVEs and latest patch version
💡 Result:
As of July 30, 2026, version 2.20.0 of jackson-databind is an older, vulnerable release [1][2]. It is affected by several security vulnerabilities discovered in 2026, including but not limited to CVE-2026-54512, CVE-2026-54513, CVE-2026-54514, and CVE-2026-59889 [3][4][5][6]. These vulnerabilities generally involve issues such as polymorphic type validation bypasses, SSRF via eager DNS resolution, and improper handling of @JsonView with @JsonUnwrapped [3][4][5][6]. The latest stable patch versions for the 2.x line are significantly newer than 2.20.0 [7][8]. As of July 2026, users should upgrade to the latest available minor/patch versions, such as 2.22.1 (released July 2026) [7]. It is strongly recommended to use the latest released version of the library to ensure all security patches are included [9][8].
Citations:
- 1: https://guide.sonatype.com/component/maven/com.fasterxml.jackson.core%3Ajackson-databind/2.20.0/vulnerabilities
- 2: https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-databind
- 3: https://osv.dev/vulnerability/CVE-2026-54512
- 4: https://osv.dev/vulnerability/CVE-2026-54513
- 5: GHSA-hgj6-7826-r7m5
- 6: https://nvd.nist.gov/vuln/detail/CVE-2026-59889
- 7: https://releasealert.dev/github/FasterXML/jackson-databind
- 8: https://github.com/FasterXML/jackson/blob/49bbda35/README.md
- 9: https://github.com/FasterXML/jackson-databind?tab=readme-ov-file
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'Files mentioning jackson:\n'
rg -n "jackson|ObjectMapper|JavaTimeModule|writeValueAsBytes|deserialize|readValue|createObjectMapper" . \
-g '!**/build/**' -g '!**/.gradle/**' -g '!**/target/**' -g '!**/node_modules/**' || true
printf '\nCandidate version catalog files:\n'
fd -a '.*\.toml$|version.*\.properties$|libs\.(gradle|kts)$|build\.gradle(\.kts)?$' . | sed 's#^\./##' | head -n 120
printf '\nRelevant MockRuleController excerpt:\n'
sed -n '1,90p' alerting-system/rest/src/main/java/ir/pathlens/alerting/rest/controller/MockRuleController.java 2>/dev/null || trueRepository: alishazaee/PathLens
Length of output: 7708
Use the version catalog for jackson-databind.
At alerting-system/rest/build.gradle and alerting-system/rest/build.gradle.kts, ObjectMapper depends on the project Jackson dependency, not the pinned 2.20.0 used by evaluator. Align these Jackson versions across modules and update them beyond vulnerable 2.15.x/2.20.x releases so the REST mock is not left on an older Jackson line.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@alerting-system/rest/src/main/java/ir/pathlens/alerting/rest/controller/MockRuleController.java`
around lines 3 - 4, Update the Jackson dependencies in
alerting-system/rest/build.gradle and build.gradle.kts to use the
version-catalog Jackson coordinates rather than hardcoded versions, and align
ObjectMapper/JavaTimeModule with the catalog-managed version. Ensure the shared
Jackson version is upgraded beyond the vulnerable 2.15.x/2.20.x lines across the
affected modules.
| public class MockRuleController implements AutoCloseable { | ||
|
|
||
| private final HttpServer server; | ||
| private final List<Rule> rules = new CopyOnWriteArrayList<>(); | ||
| private final AtomicInteger revisionNumber = new AtomicInteger(0); | ||
| private final ObjectMapper objectMapper; | ||
|
|
||
| public MockRuleController(int port) throws IOException { | ||
| objectMapper = new ObjectMapper(); | ||
| objectMapper.registerModule(new JavaTimeModule()); | ||
| server = HttpServer.create(new InetSocketAddress(port), 0); | ||
| server.createContext(ApiPathConstants.GET_ACTIVE_RULES_PATH, this::handleGetActiveRules); | ||
| server.createContext(ApiPathConstants.GET_RULES_REVISION_PATH, this::handleGetRevision); | ||
| server.setExecutor(null); | ||
| server.start(); | ||
| } | ||
|
|
||
| public MockRuleController() throws IOException { | ||
| this(0); | ||
| } | ||
|
|
||
| public void addRule(UUID id, String title, String geometryWkt, LocalDateTime expiresAt, | ||
| IdentityWrapper identity, RuleType ruleType) { | ||
| rules.add(new Rule(id, title, geometryWkt, expiresAt, identity, true, ruleType, false, LocalDateTime.now())); | ||
| revisionNumber.incrementAndGet(); | ||
| } | ||
|
|
||
| public void clearRules() { | ||
| rules.clear(); | ||
| revisionNumber.incrementAndGet(); | ||
| } | ||
|
|
||
| public String getBaseUrl() { | ||
| return "http://localhost:" + server.getAddress().getPort(); | ||
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
| server.stop(0); | ||
| } | ||
|
|
||
| private void handleGetActiveRules(HttpExchange exchange) throws IOException { | ||
| byte[] response = objectMapper.writeValueAsBytes(rules); | ||
| exchange.getResponseHeaders().set("Content-Type", "application/json"); | ||
| exchange.sendResponseHeaders(200, response.length); | ||
| try (OutputStream os = exchange.getResponseBody()) { | ||
| os.write(response); | ||
| } | ||
| } | ||
|
|
||
| private void handleGetRevision(HttpExchange exchange) throws IOException { | ||
| byte[] response = Integer.toString(revisionNumber.get()).getBytes(); | ||
| exchange.getResponseHeaders().set("Content-Type", "application/json"); | ||
| exchange.sendResponseHeaders(200, response.length); | ||
| try (OutputStream os = exchange.getResponseBody()) { | ||
| os.write(response); | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy lift
Test-only mock server lives in the rest module's production sources.
MockRuleController is a test double (per EvaluatorTest.java, it's only instantiated in test setup) but it's placed under alerting-system/rest/src/main/java/..., so it ships inside the rest module's production artifact. The repo already has a dedicated common-libs/test-extensions module for cross-module test support (see KafkaExtension.java in this same PR) — this class would fit better there or in a rest test-fixtures source set, keeping mock/test-only HTTP servers out of the shipped production jar.
Confirms the evaluator runtime can discover the mock controller REST base URL (including ephemeral-port binding) before rule fetching begins. which shows this is purely test wiring, not a production dependency of rest.
🧰 Tools
🪛 ast-grep (0.45.0)
[warning] 55-55: Avoid building a URL host from untrusted input
Context: "http://localhost:" + server.getAddress().getPort()
Note: [CWE-20] Improper Input Validation.
(tainted-url-host)
[warning] 73-73: Use a randomly-generated IV
Context: byte[] response = Integer.toString(revisionNumber.get()).getBytes();
Note: [CWE-329] Generation of Predictable IV with CBC Mode.
(random-iv)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@alerting-system/rest/src/main/java/ir/pathlens/alerting/rest/controller/MockRuleController.java`
around lines 23 - 81, Move the test-only MockRuleController class out of the
rest module’s production sources into the shared test-extensions module or the
rest module’s test-fixtures source set. Update evaluator test setup and any
imports or dependencies to use the relocated class, preserving its REST
endpoints and ephemeral-port base URL behavior.
| while (!polledRecords.offer(record)) { | ||
| if (!running) { | ||
| break; | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Exit the batch loop when shutdown is observed.
break exits only the inner retry loop. The enclosing for then continues tracking and potentially enqueueing remaining records after close() has set running to false. Return from pollLoop (the finally block will still run) when shutdown is observed.
Proposed fix
for (ConsumerRecord<K, V> record : records) {
+ if (!running) {
+ return;
+ }
TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());
tracker.track(new OffsetPartition(topicPartition, record.offset()));
while (!polledRecords.offer(record)) {
if (!running) {
- break;
+ return;
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| while (!polledRecords.offer(record)) { | |
| if (!running) { | |
| break; | |
| } | |
| for (ConsumerRecord<K, V> record : records) { | |
| if (!running) { | |
| return; | |
| } | |
| TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition()); | |
| tracker.track(new OffsetPartition(topicPartition, record.offset())); | |
| while (!polledRecords.offer(record)) { | |
| if (!running) { | |
| return; | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java`
around lines 104 - 107, Update the record-processing loop in
KafkaParallelConsumer.pollLoop so observing running == false while retrying
polledRecords.offer(record) exits pollLoop immediately rather than only breaking
the inner retry loop; return from the method so its existing finally block still
executes.
|
|
||
| @Override | ||
| public void beforeEach(ExtensionContext extensionContext) throws Exception { | ||
| Set<String> topics = listTopics(); | ||
| extensionContext.getStore(ExtensionContext.Namespace.create(getClass())) | ||
| .put(TOPICS_KEY, topics); | ||
| } | ||
|
|
||
| @Override | ||
| public void afterEach(ExtensionContext extensionContext) throws Exception { | ||
| Set<String> preExistingTopics = extensionContext.getStore(ExtensionContext.Namespace.create(getClass())) | ||
| .get(TOPICS_KEY, Set.class); | ||
| if (preExistingTopics != null) { | ||
| deleteNewTopics(preExistingTopics); | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
does this project's gradle/junit-platform config enable JUnit 5 parallel test execution
💡 Result:
Whether parallel execution is enabled depends on how you configure it, as there are two distinct, independent mechanisms for parallelizing tests in a Gradle project using JUnit 5 [1]. 1. JUnit 5 Parallel Execution (Internal) This mechanism executes tests in parallel within a single JVM process [2][3]. To enable this, you must configure JUnit 5-specific properties [2][4]. Simply calling useJUnitPlatform in your Gradle build script does not enable this [5][6]. You can enable it by creating a file named junit-platform.properties in your src/test/resources directory containing [2][7][8]: junit.jupiter.execution.parallel.enabled = true Alternatively, you can pass these configuration parameters directly in your build.gradle file using system properties or the configurationParameters DSL [9][5][6]: test { useJUnitPlatform systemProperty("junit.jupiter.execution.parallel.enabled", "true") // Additional settings like mode and strategy are required to actually run tests in parallel [2][3] } 2. Gradle Test Task Parallelism (Process-based) Gradle can also run tests in parallel by forking multiple JVM processes [1]. This is independent of JUnit 5's internal parallel execution [1]. You enable this by setting the maxParallelForks property in your build script [1][10]: test { useJUnitPlatform maxParallelForks = 4 // Runs up to 4 test classes in parallel processes [1] } Summary If you want JUnit 5 to manage parallel execution (e.g., parallelizing test methods within the same class), you must set junit.jupiter.execution.parallel.enabled=true via a properties file or system property [2][3]. If you only want to run different test classes in parallel, configuring Gradle's maxParallelForks is typically sufficient and simpler [1]. These two approaches can be combined, but doing so requires careful management of shared resources to avoid intermittent failures [1].
Citations:
- 1: https://docs.gradle.org/current/userguide/java_testing.html
- 2: https://docs.junit.org/6.1.2/writing-tests/parallel-execution.html
- 3: https://docs.junit.org/5.14.2/writing-tests/parallel-execution
- 4: https://docs.junit.org/6.0.3/writing-tests/parallel-execution.html
- 5: https://docs.junit.org/snapshot/running-tests/build-support
- 6: https://docs.junit.org/5.14.2/running-tests/build-support.html
- 7: https://www.baeldung.com/junit-5-parallel-tests
- 8: https://www.javacodegeeks.com/junit5-gradle-parallel-testing-example.html
- 9: Support JUnit platform configuration parameters gradle/gradle#7622
- 10: https://www.baeldung.com/junit5-gradle-parallel-testing
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== locate relevant files =="
git ls-files | rg '(^|/)(build\.gradle(\.kts)?|gradle\.properties|junit-platform\.properties|KafkaExtension\.java)$|common-libs/test-extensions/src/main/java/ir/pathlens/extension/kafka/KafkaExtension\.java'
echo
echo "== KafkaExtension outline =="
ast-grep outline common-libs/test-extensions/src/main/java/ir/pathlens/extension/kafka/KafkaExtension.java --view expanded || true
echo
echo "== KafkaExtension contents =="
cat -n common-libs/test-extensions/src/main/java/ir/pathlens/extension/kafka/KafkaExtension.java
echo
echo "== JUnit/Gradle parallel config references =="
rg -n --hidden --glob '!build/**' --glob '!**/.gradle/**' \
'junit\.jupiter\.execution\.parallel\.enabled|maxParallelForks|junit-platform\.properties|ConfigurationParameters|System\.setProperty|ParallelExecutionConfig|JUnitPlatformOptions' . || true
echo
echo "== dependency mentions for Kafka and JUnit/TestContainers =="
for f in $(git ls-files | rg '(^|/)(build\.gradle(\.kts)?|gradle\.properties|pom\.xml)$' | head -50); do
echo "--- $f"
rg -n 'kafkacontainer|testcontainers|junit-jupiter|junit-platform|kafka-avro-serializer|kafka-clients' "$f" || true
doneRepository: alishazaee/PathLens
Length of output: 6201
Don’t clean Kafka topics from a shared beforeEach/afterEach snapshot.
KafkaExtension keeps one KafkaContainer for the JVM and stores each test’s per-test snapshot in extensionContext.getStore(...). Under JUnit5 JUnitPlatform or Gradle parallel execution, concurrently-running test classes can share the same Kafka server; one test’s afterEach can compute topicsToDelete against its own before-snapshot and delete topics created by another concurrent test. Fix this before enabling parallel tests, or isolate each concurrent execution context/test class.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@common-libs/test-extensions/src/main/java/ir/pathlens/extension/kafka/KafkaExtension.java`
around lines 38 - 53, Update KafkaExtension’s beforeEach/afterEach topic-cleanup
design so concurrent tests sharing the JVM KafkaContainer cannot delete topics
created by other executions; replace the per-test snapshot deletion in
beforeEach, afterEach, and deleteNewTopics with execution isolation or
synchronization that protects each test class/context’s topics. Preserve cleanup
of topics created by the current execution while preventing cross-test deletion
during parallel runs.
1 participant
Summary by CodeRabbit