sv2000 commented on a change in pull request #3214:
URL: https://github.com/apache/incubator-gobblin/pull/3214#discussion_r565644740
##########
File path:
gobblin-core-base/src/main/java/org/apache/gobblin/source/extractor/extract/FlushingExtractor.java
##########
@@ -108,8 +108,8 @@
protected Long flushIntervalMillis;
protected Long timeOfLastFlush = System.currentTimeMillis();
- private FlushAckable lastFlushAckable;
- private boolean hasOutstandingFlush = false;
+ protected FlushAckable lastFlushAckable;
Review comment:
See the comment below about creating a shutdown() method in
FlushingExtractor class.
##########
File path:
gobblin-runtime/src/main/java/org/apache/gobblin/runtime/StreamModelTaskRunner.java
##########
@@ -89,6 +89,7 @@ protected void run() throws Exception {
// by publish, so set the shutdownRequested flag on cancel to stop the
extractor
Flowable streamWithShutdownOnCancel = connectableStream.doOnCancel(() -> {
this.shutdownRequested.set(true);
+ // In the case that extractor sucks in reading record when cancel get
called, we call shutdown again to force it
Review comment:
sucks -> stuck.
##########
File path:
gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/source/extractor/extract/kafka/KafkaStreamingExtractor.java
##########
@@ -139,6 +129,10 @@ public void shutdown() {
log.error("Interrupted when attempting to shutdown metrics collection
threads.");
}
this.shutdownRequested.set(true);
+ // In case hasOutstandingFlush, we need to manually nack the ackable to
make sure the CountDownLatch not hang
Review comment:
Can we move this block to the parent class inside its own shutdown()
method and call super.shutdown() here?
----------------------------------------------------------------
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]
