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]
