THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
Code Block | ||||
---|---|---|---|---|
| ||||
WIP/** * Startpoint represents a position in a stream partition. */ public abstract class Startpoint { private final long creationTimestamp; /** * Applies the {@link StartpointVisitor}'s visit methods to the {@link Startpoint} * and returns the result of that operation. * @param input the metadata associated with the startpoint. * @param startpointVisitor the visitor of the startpoint. * @return the result of applying the visitor on startpoint. */ public abstract <IN, OUT> OUT apply(IN input, StartpointVisitor<IN, OUT> startpointVisitor); } /** * A {@link Startpoint} that represents the earliest offset in a stream partition. */ public final class StartpointOldest extends Startpoint { @Override public <IN, OUT> OUT apply(IN input, StartpointVisitor<IN, OUT> startpointVisitor) { return startpointVisitor.visit(input, this); } } /** * A {@link Startpoint} that represents the latest offset in a stream partition. */ public final class StartpointUpcoming extends Startpoint { @Override public <IN, OUT> OUT apply(IN input, StartpointVisitor<IN, OUT> startpointVisitor) { return startpointVisitor.visit(input, this); } } /** * A {@link Startpoint} that represents a timestamp offset in a stream partition. */ public final class StartpointTimestamp extends Startpoint { private final Long timestampOffset; } |
StartpointVisitor
Code Block | ||||
---|---|---|---|---|
| ||||
WIP/** * A {@link SystemAdmin} implementation should implement this abstraction to support {@link Startpoint}. */ public interface StartpointVisitor<IN, OUT> { /** * Performs a sequence of operations using the {@link IN} and {@link StartpointSpecific} and returns the result of the execution. * @param input the input metadata about the startpoint. * @param startpointSpecific the {@link Startpoint} that represents the specific offset. * @return the result of executing the operations defined by the visit method. */ default OUT visit(IN input, StartpointSpecific startpointSpecific) { throw new UnsupportedOperationException("StartpointSpecific is not supported."); } /** * Performs a sequence of operations using the {@link IN} and {@link StartpointTimestamp} and returns the result of the execution. * @param input the input metadata about the startpoint. * @param startpointTimestamp the {@link Startpoint} that represents the timestamp. * @return the result of executing the operations defined by the visit method. */ default OUT visit(IN input, StartpointTimestamp startpointTimestamp) { throw new UnsupportedOperationException("StartpointTimestamp is not supported."); } /** * Performs a sequence of operations using the {@link IN} and {@link StartpointOldest} and returns the result of the execution. * @param input the input metadata about the startpoint. * @param startpointOldest the {@link Startpoint} that represents the earliest offset. * @return the result of executing the operations defined by the visit method. */ default OUT visit(IN input, StartpointOldest startpointOldest) { throw new UnsupportedOperationException("StartpointOldest is not supported."); } /** * Performs a sequence of operations using the {@link IN} and {@link StartpointUpcoming} and returns the result of the execution. * @param input the input metadata about the startpoint. * @param startpointUpcoming the {@link Startpoint} that represents the latest offset. * @return the result of executing the operations defined by the visit method. */ default OUT visit(IN input, StartpointUpcoming startpointUpcoming) { throw new UnsupportedOperationException("StartpointUpcoming is not supported."); } } |
Storing Requested Startpoint
...