cameronlee314 commented on a change in pull request #940: SAMZA-2121: Add
checkpoint offset field to IncomingMessageEnvelope
URL: https://github.com/apache/samza/pull/940#discussion_r263639549
##########
File path:
samza-api/src/main/java/org/apache/samza/system/IncomingMessageEnvelope.java
##########
@@ -86,7 +101,29 @@ public IncomingMessageEnvelope(SystemStreamPartition
systemStreamPartition, Stri
*/
public IncomingMessageEnvelope(SystemStreamPartition systemStreamPartition,
String offset,
Object key, Object message, int size, long eventTime, long arrivalTime) {
- this(systemStreamPartition, offset, key, message, size);
+ this(systemStreamPartition, offset, offset, key, message, size, eventTime,
arrivalTime);
+ }
+
+ /**
+ * Constructs a new IncomingMessageEnvelope from specified components
+ * @param systemStreamPartition The aggregate object representing the
incoming stream name, the name of the cluster
+ * from which the stream came, and the partition of the stream from which
the message was received.
+ * @param offset The offset in the partition that the message was received
from.
+ * @param checkpointOffset offset that can be checkpointed when this {@link
IncomingMessageEnvelope} is processed
+ * @param key A deserialized key received from the partition offset.
+ * @param message A deserialized message received from the partition offset.
+ * @param size size of the message and key in bytes.
+ * @param eventTime the timestamp (in epochMillis) of when this event
happened
+ * @param arrivalTime the timestamp (in epochMillis) of when this event
arrived to (i.e., was picked-up by) Samza
+ */
+ public IncomingMessageEnvelope(SystemStreamPartition systemStreamPartition,
String offset, String checkpointOffset,
+ Object key, Object message, int size, long eventTime, long arrivalTime) {
+ this.systemStreamPartition = systemStreamPartition;
+ this.offset = offset;
+ this.checkpointOffset = checkpointOffset;
Review comment:
1. Some more details and an example are described at
https://issues.apache.org/jira/browse/SAMZA-2120.
2. We could make this part of the internal API by creating an
`InternalIncomingMessageEnvelope` which is passed around internally and extends
`IncomingMessageEnvelope`. Would that help avoid complicating the public API?
----------------------------------------------------------------
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:
us...@infra.apache.org
With regards,
Apache Git Services
