[
https://issues.apache.org/jira/browse/BEAM-12333?focusedWorklogId=598061&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-598061
]
ASF GitHub Bot logged work on BEAM-12333:
-----------------------------------------
Author: ASF GitHub Bot
Created on: 17/May/21 18:28
Start Date: 17/May/21 18:28
Worklog Time Spent: 10m
Work Description: xinyuiscool commented on a change in pull request
#14802:
URL: https://github.com/apache/beam/pull/14802#discussion_r633684578
##########
File path:
runners/samza/src/main/java/org/apache/beam/runners/samza/runtime/SamzaTimerInternalsFactory.java
##########
@@ -702,8 +683,18 @@ public void encode(TimerKey<K> value, OutputStream
outStream)
if (keyCoder != null) {
key = keyCoder.decode(inStream);
}
-
- return new TimerKey<>(key, namespace, timerId);
+
timerKeyBuilder.setTimerId(timerId).setStateNamespace(namespace).setKey(key);
+
+ // check if the stream has more available bytes. This is to ensure
backward compatibility with
+ // old rocksdb state
+ // which does not encode timer family data
+ if (inStream.available() > 0 && BOOLEAN_CODER.decode(inStream)) {
Review comment:
As commented above, we can ignore this bool flag.
##########
File path:
runners/samza/src/main/java/org/apache/beam/runners/samza/runtime/SamzaTimerInternalsFactory.java
##########
@@ -682,16 +653,26 @@ public void encode(TimerKey<K> value, OutputStream
outStream)
throws CoderException, IOException {
// encode the timestamp first
- STRING_CODER.encode(value.timerId, outStream);
- STRING_CODER.encode(value.stateNamespace.stringKey(), outStream);
+ STRING_CODER.encode(value.getTimerId(), outStream);
+ STRING_CODER.encode(value.getStateNamespace().stringKey(), outStream);
if (keyCoder != null) {
- keyCoder.encode(value.key, outStream);
+ keyCoder.encode(value.getKey(), outStream);
+ }
+
+ // encode a byte (1/0) to indicate the presence/absence of timerFamilyId
+ // We can use this approach to add additional fields in the future
+ if (!Strings.isNullOrEmpty(value.getTimerFamilyId())) {
+ BOOLEAN_CODER.encode(true, outStream);
Review comment:
@ajothomas : I think we should be able to blindly encode timeFamilyId as
it's going to be defaulted as blank "". See TimerInternals: line 198. The new
TimerDataCoder (TimerDataCoderV2) also encode it directly without the need to
do null check. We should be able to follow that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 598061)
Time Spent: 1h 10m (was: 1h)
> TimerFamilyId is not present in TimerKey in SamzaTimerInternalsFactory and
> therefore doesn't get persisted for Processing Time timers
> -------------------------------------------------------------------------------------------------------------------------------------
>
> Key: BEAM-12333
> URL: https://issues.apache.org/jira/browse/BEAM-12333
> Project: Beam
> Issue Type: Improvement
> Components: runner-samza
> Reporter: Ajo Thomas
> Assignee: Ajo Thomas
> Priority: P3
> Time Spent: 1h 10m
> Remaining Estimate: 0h
>
> `SamzaTimerInternalsFactory` stores timer state, event time and processing
> time timers alike, in the form of a MapState<TimerKey<K>, Long>. TimerKey,
> however, doesn't include the `TimerFamilyId` and is therefore not persisted.
> On the other hand, event time timers uses an additional time sorted set to
> persist the TimerData(which contains TimerFamilyId). We need to include timer
> family id in the TimerKey to ensure that it is persisted.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
