shanthoosh commented on a change in pull request #918: SAMZA-2094: Implement
the StartpointVisitor for the KafkaSystemConsumer.
URL: https://github.com/apache/samza/pull/918#discussion_r258333602
##########
File path:
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemConsumer.java
##########
@@ -258,40 +260,28 @@ public void stop() {
*/
@Override
public void register(SystemStreamPartition systemStreamPartition, String
offset) {
+ register(systemStreamPartition, new StartpointSpecific(offset));
+ }
+
+ @Override
+ public void register(SystemStreamPartition systemStreamPartition, Startpoint
startpoint) {
if (started.get()) {
- String msg = String.format("%s: Trying to register partition after
consumer has been started. ssp=%s", this,
- systemStreamPartition);
- throw new SamzaException(msg);
+ String exceptionMessage = String.format("KafkaSystemConsumer: %s had
started. Registration of ssp: %s, startpoint: %s failed.", this,
systemStreamPartition, startpoint);
Review comment:
This was not introduced as a part of this patch. However, I agree that this
is unnecessary. Fixed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
