Thanks Chesnay!
I tried giving that a shot but I still wasn't able to access the
globalJobParameters from within the open function in my KeyedProcessFunction.
You can see the implementation below which I believe should be correct:
object CustomProcessFunctionTestHarness {
fun <K, IN, OUT> forKeyedProcessFunction(
function: KeyedProcessFunction<K, IN, OUT>,
keySelector: KeySelector<IN, K>,
keyType: TypeInformation<K>,
parameters: ParameterTool
): KeyedOneInputStreamOperatorTestHarness<K, IN, OUT> {
val testHarness: KeyedOneInputStreamOperatorTestHarness<K, IN, OUT>
=
KeyedOneInputStreamOperatorTestHarness<K, IN, OUT>(
KeyedProcessOperator<K, IN,
OUT>(Preconditions.checkNotNull(function)),
keySelector,
keyType,
1,
1,
0
)
// Adjust execution configuration via parameters
testHarness.executionConfig.globalJobParameters = parameters
testHarness.open()
return testHarness
}
}
Usage-wise, the declaration looks as you might expect:
bufferedMagicWindowHarness =
CustomProcessFunctionTestHarness.forKeyedProcessFunction(
MagicWindowFunction(),
{ log -> log.getKey() },
TypeInformation.of(String::class.java),
ParameterTool.fromArgs(arrayOf("--processingTimeBuffer",
"60000"))
)
And then I think as I described earlier, these are attempted to be read via
the following in the open() function:
val parameters = runtimeContext.executionConfig.globalJobParameters as?
ParameterTool
if (parameters != null) {
processingTimeBuffer = parameters.getLong("processingTimeBuffer")
}
Does anything look out of place here? I haven't gone spelunking into the
source code for this yet, but I'm assuming that I'm setting the correct
values on the execution configuration.
Thanks again,
Rion
On Thu, Mar 4, 2021 at 7:57 AM Chesnay Schepler <ches...@apache.org> wrote:
> The reason why your attempts have failed is that
> ProcessFunctionTestHarnesses.forKeyedProcessFunction automatically calls
> open(), thus any mutations on the harness happen too late.
>
> I'd suggest to take a look at the implementation of that method and
> essentially copy the code.
> You can then call the harness constructor manually and mutate the
> execution config before calling open().
>
> On 3/4/2021 2:49 PM, Rion Williams wrote:
>
> Absolutely,
>
> I think it's gone through quite a few iterations, but this is the current
> state of it (defined in a @Before function as part of scaffolding out the
> tests):
>
> private lateinit var magicWindowHarness:
> KeyedOneInputStreamOperatorTestHarness<String, Log, FileOutput>
>
> @Before
> fun init() {
> magicWindowHarness =
> ProcessFunctionTestHarnesses.forKeyedProcessFunction(
> MagicWindowFunction(),
> { log -> log.getKey() },
> TypeInformation.of(String::class.java)
> )
> }
>
> I've also tried a few variants of that with a separate declaration for the
> function itself, etc.
>
>
>
> On Thu, Mar 4, 2021 at 6:47 AM Chesnay Schepler <ches...@apache.org>
> wrote:
>
>> Could you show us how you create test harness?
>>
>> On 3/4/2021 5:13 AM, Rion Williams wrote:
>>
>> Hi all,
>>
>> Early today I had asked a few questions regarding the use of the many
>> testing constructs available within Flink and believe that I have things in
>> a good direction at present. I did run into a specific case that either may
>> not be supported, or just isn't documented well enough for me to determine
>> what is going wrong.
>>
>> Basically, I have a KeyedProcessFunction that reads some global-level
>> configuration via GlobalJobParameters within its open function:
>>
>> override fun open(configuration: Configuration) {
>> // Omitted for brevity
>>
>> val parameters = runtimeContext.executionConfig.globalJobParameters
>> as? ParameterTool
>> if (parameters != null) {
>> processingTimeBuffer = parameters.getLong("processingTimeBuffer",
>> 0L)
>> }
>> }
>>
>> This works just as expected within the actual pipeline itself when set
>> similarly:
>>
>> streamEnvironment.config.globalJobParameters = parameters
>>
>> However, I don't see an effective way to set this against a TestHarness
>> as I've made several attempts but I never can seem to populate the
>> globalJobParameters property within the KeyedProcessFunction itself using a
>> test harness despite multiple attempts
>>
>> // Attempt 1
>> magicWindowHarness.operator.runtimeContext.executionConfig.globalJobParameters
>> = ParameterTool.fromMap(...)
>>
>> // Attempt 2
>> magicWindowHarness.executionConfig.globalJobParameters =
>> ParameterTool.fromMap(...)
>>
>> // Attempt 3
>> magicWindowHarness.environment.executionConfig.globalJobParameters =
>> ParameterTool.fromMap(...)
>>
>> // Attempt 4
>> val env = StreamExecutionEnvironment.
>> env.config.globalJobParameters = ParameterTool.fromMap(...)
>>
>> Is this supported or am I simply going about it the wrong way? Or even
>> just perhaps missing a piece of the puzzle?
>>
>> Thanks much,
>>
>> Rion
>>
>>
>>
>
