Fix scenario tests with fragment input validation #7159
Conversation
|
Caution Review failedThe pull request is closed. 📝 Walkthrough📝 WalkthroughWalkthroughThe pull request introduces enhancements to the testing framework for fragments with parameter validation in the Changes
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
8a7be19 to
c7832ab
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Outside diff range and nitpick comments (2)
engine/flink/executor/src/test/scala/pl/touk/nussknacker/engine/process/runner/FlinkTestMainSpec.scala (2)
667-682: Enhance the test with specific assertions for fragment behaviorWhile the test verifies that no exceptions are thrown when a process fragment with parameter validation is defined, it would be more robust to include assertions that check the actual behavior of the fragment. This ensures that not only does the process run without exceptions, but it also produces the expected results.
739-766: Consolidateobjectandclassto improve code organizationDefining an
object FlinkTestMainSpecseparately in the same file asclass FlinkTestMainSpeccan be confusing. Consider moving the contents of theobjectinto theclassasprivatemembers. This aligns with Scala best practices and enhances readability.Apply this diff to move the definitions into the class:
-class FlinkTestMainSpec extends AnyWordSpec with Matchers with Inside with BeforeAndAfterEach with OptionValues { +class FlinkTestMainSpec extends AnyWordSpec with Matchers with Inside with BeforeAndAfterEach with OptionValues { + private val fragmentWithValidationName = "fragmentWithValidation" + + private val processWithFragmentParameterValidation: CanonicalProcess = { + // existing code from the object + } + // Rest of the class code... -} - -object FlinkTestMainSpec { - private val fragmentWithValidationName = "fragmentWithValidation" - - private val processWithFragmentParameterValidation: CanonicalProcess = { - // existing code... - } -}
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (2)
engine/flink/executor/src/test/scala/pl/touk/nussknacker/engine/process/runner/FlinkTestMainSpec.scala(4 hunks)utils/utils/src/main/scala/pl/touk/nussknacker/engine/util/json/ToJsonEncoder.scala(1 hunks)
✅ Files skipped from review due to trivial changes (1)
- utils/utils/src/main/scala/pl/touk/nussknacker/engine/util/json/ToJsonEncoder.scala
...nk/executor/src/test/scala/pl/touk/nussknacker/engine/process/runner/FlinkTestMainSpec.scala
Show resolved
Hide resolved
...nk/executor/src/test/scala/pl/touk/nussknacker/engine/process/runner/FlinkTestMainSpec.scala
Show resolved
Hide resolved
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Outside diff range and nitpick comments (1)
scenario-compiler/src/main/scala/pl/touk/nussknacker/engine/compile/ProcessCompilerData.scala (1)
43-43: Document performance implications of using unoptimized evaluator.While switching to
unOptimizedEvaluatormay resolve the serialization issue in tests, it's important to document this decision as it represents a performance trade-off.Add a comment explaining why we're using the unoptimized evaluator:
val expressionEvaluator = + // Using unoptimized evaluator to avoid serialization issues in tests ExpressionEvaluator.unOptimizedEvaluator(globalVariablesPreparer)
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (2)
engine/flink/executor/src/test/scala/pl/touk/nussknacker/engine/process/runner/FlinkTestMainSpec.scala(4 hunks)scenario-compiler/src/main/scala/pl/touk/nussknacker/engine/compile/ProcessCompilerData.scala(1 hunks)
🔇 Additional comments (2)
scenario-compiler/src/main/scala/pl/touk/nussknacker/engine/compile/ProcessCompilerData.scala (1)
43-43: Verify that this change resolves the serialization issue.
The switch to unOptimizedEvaluator and removal of listeners parameter might help with serialization, but we should verify this thoroughly.
Let's verify the impact:
✅ Verification successful
Based on the search results, I can now provide a definitive response regarding the switch to unoptimized evaluator:
The change to unoptimized evaluator is appropriate for test scenarios
The switch from optimizedEvaluator to unOptimizedEvaluator and removal of listeners is justified because:
- This is in test context where performance is less critical
- The change is consistent with other test-related code that uses unoptimized evaluator (as seen in multiple test files)
- The removal of listeners helps with serialization which is important for Flink test execution
- The change is localized to test scenarios with fragment input validation
The evidence shows this is a valid change that prioritizes test stability over performance optimization, which is the right tradeoff for test scenarios. The unoptimized evaluator is already used in other test contexts throughout the codebase.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Search for related test cases and their outcomes
rg -A 5 "RichFlatMapFunction.*serializ"
# Look for other uses of optimizedEvaluator to ensure we haven't missed any cases
ast-grep --pattern 'ExpressionEvaluator.optimizedEvaluator($$$)'
Length of output: 338
Script:
#!/bin/bash
# Let's check for RichFlatMapFunction usage and serialization context
rg -B 5 -A 5 "RichFlatMapFunction"
# Let's also check where unOptimizedEvaluator is used
ast-grep --pattern 'ExpressionEvaluator.unOptimizedEvaluator($$$)'
# Check for any serialization-related tests or issues
rg -B 3 -A 3 "NotSerializableException|Serializable"
Length of output: 266090
engine/flink/executor/src/test/scala/pl/touk/nussknacker/engine/process/runner/FlinkTestMainSpec.scala (1)
140-161: Add documentation and improve type safety
The fragment definition could benefit from:
- ScalaDoc explaining the purpose and usage of this test fragment
- More type-safe alternatives for empty collections
Consider these improvements:
+ /**
+ * Creates a test fragment with parameter validation.
+ * This fragment is used to verify that the parameter validation mechanism
+ * works correctly in the context of fragment processing.
+ */
private val processWithFragmentParameterValidation: CanonicalProcess = {
val fragmentParamName = ParameterName("param")
val fragmentParam = FragmentParameter(fragmentParamName, FragmentClazzRef[String]).copy(
valueCompileTimeValidation = Some(
ParameterValueCompileTimeValidation(
validationExpression = Expression.spel("true"),
validationFailedMessage = Some("param validation failed")
)
)
)
CanonicalProcess(
MetaData(fragmentWithValidationName, FragmentSpecificData()),
List(
FlatNode(
- FragmentInputDefinition("start", List(fragmentParam))
+ FragmentInputDefinition("start", fragmentParam :: Nil)
),
- FlatNode(FragmentOutputDefinition("out1", "output", List.empty))
+ FlatNode(FragmentOutputDefinition("out1", "output", Nil))
),
- List.empty
+ Nil
)
}Additionally, according to the PR objectives, there's a serialization issue with RichFlatMapFunction. However, I don't see any @transient annotations being added to address this. Could you clarify if this is being handled in a different file?
Let me check for any related changes:
...nk/executor/src/test/scala/pl/touk/nussknacker/engine/process/runner/FlinkTestMainSpec.scala
Outdated
Show resolved
Hide resolved
dd3b6c1 to
5a0e87d
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (2)
engine/flink/executor/src/test/scala/pl/touk/nussknacker/engine/process/runner/FlinkTestMainSpec.scala(6 hunks)scenario-compiler/src/main/scala/pl/touk/nussknacker/engine/compile/ProcessCompilerData.scala(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- scenario-compiler/src/main/scala/pl/touk/nussknacker/engine/compile/ProcessCompilerData.scala
🔇 Additional comments (1)
engine/flink/executor/src/test/scala/pl/touk/nussknacker/engine/process/runner/FlinkTestMainSpec.scala (1)
808-829: 🛠️ Refactor suggestion
Provide a meaningful validation expression
The current validation expression Expression.spel("true") will always pass validation, making it ineffective for testing the validation mechanism. Consider using an expression that actually validates the parameter value.
Apply this diff to use a meaningful validation expression:
val processWithFragmentParameterValidation: CanonicalProcess = {
val fragmentParamName = ParameterName("param")
val fragmentParam = FragmentParameter(fragmentParamName, FragmentClazzRef[String]).copy(
valueCompileTimeValidation = Some(
ParameterValueCompileTimeValidation(
- validationExpression = Expression.spel("true"),
+ validationExpression = Expression.spel("#param != null and #param.length() > 0"),
validationFailedMessage = Some("param validation failed")
)
)
)Likely invalid or redundant comment.
...nk/executor/src/test/scala/pl/touk/nussknacker/engine/process/runner/FlinkTestMainSpec.scala
Show resolved
Hide resolved
b4c2d43 to
1d3611c
Compare
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Outside diff range and nitpick comments (1)
engine/flink/executor/src/test/scala/pl/touk/nussknacker/engine/process/runner/FlinkTestMainSpec.scala (1)
24-29: Consider consolidating imports from the same packageThe imports from
pl.touk.nussknacker.engine.api.parameterandpl.touk.nussknacker.engine.graph.nodepackages can be consolidated to improve readability.-import pl.touk.nussknacker.engine.api.parameter.ParameterName -import pl.touk.nussknacker.engine.api.parameter.ParameterValueCompileTimeValidation +import pl.touk.nussknacker.engine.api.parameter.{ParameterName, ParameterValueCompileTimeValidation}
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (3)
docs/Changelog.md(1 hunks)engine/flink/executor/src/test/scala/pl/touk/nussknacker/engine/process/runner/FlinkTestMainSpec.scala(4 hunks)scenario-compiler/src/main/scala/pl/touk/nussknacker/engine/compile/ProcessCompilerData.scala(2 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- scenario-compiler/src/main/scala/pl/touk/nussknacker/engine/compile/ProcessCompilerData.scala
🔇 Additional comments (4)
engine/flink/executor/src/test/scala/pl/touk/nussknacker/engine/process/runner/FlinkTestMainSpec.scala (3)
668-683: 🛠️ Refactor suggestion
Enhance test coverage for fragment parameter validation
The current test only verifies the happy path. Consider:
- Testing with invalid parameters to ensure validation works
- Verifying validation error messages
- Using a more descriptive test name that indicates what validation scenario is being tested
- "should not throw exception when process fragment has parameter validation defined" in {
+ "should validate fragment parameters according to validation rules" in {
val scenario = ScenarioBuilder
.streaming("scenario1")
.source(sourceNodeId, "input")
.fragmentOneOut("sub", fragmentWithValidationName, "output", "fragmentResult", "param" -> "'asd'".spel)
.emptySink("out", "valueMonitor", "Value" -> "1".spel)
val resolved = FragmentResolver(List(processWithFragmentParameterValidation)).resolve(scenario)
- val results = runFlinkTest(
+ // Test valid parameter
+ val validResults = runFlinkTest(
resolved.valueOr { _ => throw new IllegalArgumentException("Won't happen") },
ScenarioTestData(List(ScenarioTestJsonRecord(sourceNodeId, Json.fromString("0|1|2|3|4|5|6")))),
useIOMonadInInterpreter
)
- results.exceptions.length shouldBe 0
+ validResults.exceptions shouldBe empty
+
+ // Test invalid parameter
+ val scenarioWithInvalidParam = scenario.copy(nodes = scenario.nodes.map {
+ case n if n.id == "sub" => n.copy(parameters = Map("param" -> "''".spel))
+ case n => n
+ })
+ val resolvedInvalid = FragmentResolver(List(processWithFragmentParameterValidation)).resolve(scenarioWithInvalidParam)
+ val invalidResults = runFlinkTest(
+ resolvedInvalid.valueOr { _ => throw new IllegalArgumentException("Won't happen") },
+ ScenarioTestData(List(ScenarioTestJsonRecord(sourceNodeId, Json.fromString("0|1|2|3|4|5|6")))),
+ useIOMonadInInterpreter
+ )
+ invalidResults.exceptions should not be empty
+ invalidResults.exceptions.head.getMessage should include("param validation failed")
}Likely invalid or redundant comment.
744-765:
Improve fragment validation implementation
The current implementation has several areas for improvement:
- The validation expression
Expression.spel("true")always returns true, making the validation ineffective - The validation message could be more descriptive
- Consider adding documentation for the fragment parameter type and validation rules
val fragmentParam = FragmentParameter(fragmentParamName, FragmentClazzRef[String]).copy(
valueCompileTimeValidation = Some(
ParameterValueCompileTimeValidation(
- validationExpression = Expression.spel("true"),
- validationFailedMessage = Some("param validation failed")
+ validationExpression = Expression.spel("#param != null and #param.length() > 0"),
+ validationFailedMessage = Some("Parameter 'param' must not be null or empty")
)
)
)Likely invalid or redundant comment.
675-676: Verify serialization handling in FragmentResolver
Since the original issue was related to serialization of RichFlatMapFunction, let's verify that the FragmentResolver properly handles serialization.
docs/Changelog.md (1)
103-103: LGTM: Changelog entry is accurate and well-placed.
The entry properly documents the fix for scenario tests with fragment input validation, with correct PR reference and placement in version 1.18 section.
1d3611c to
9c1054b
Compare
Describe your changes
Currently when we try to test a scenario which uses a fragment which has some validation set on parameter's expression there is an exception during serialization.
This happens as deep down there is a classloader passed which cannot be serialized.
The problem started to occur with introduction of
ValidationExpressionParameterValidatorwhich holdsExpressionEvaluatorwhich was created in an optimized way and thus contains listeners. EspeciallyResultsCollectingListenerwhich has avariableEncoderwhich has the problematic classloader.In proposed changes at the level of
ProcessCompilerDataI passExpressionEvaluator.unOptimizedEvaluatorwhich is used only by the validator.I also left the
ExpressionEvaluator.optimizedEvaluatorintact for theInterpreterandProcessCompilerDatacreation as it's needed there.I added a test replicating the issue. It passes with the proposed change but would fail in the same way as on the environment without it:
Checklist before merge
Summary by CodeRabbit
Summary by CodeRabbit
New Features
Bug Fixes
Documentation