echauchot commented on code in PR #3:
URL:
https://github.com/apache/flink-connector-cassandra/pull/3#discussion_r1105844502
##########
flink-connector-cassandra/src/main/java/org/apache/flink/connector/cassandra/source/enumerator/CassandraSplitEnumerator.java:
##########
@@ -57,7 +57,22 @@ public CassandraSplitEnumerator(
@Override
public void handleSplitRequest(int subtaskId, @Nullable String
requesterHostname) {
- assignUnprocessedSplitToReader(subtaskId);
+ checkReaderRegistered(subtaskId);
+ final CassandraSplit cassandraSplit = state.getASplit();
+ if (cassandraSplit != null) {
+ LOG.info("Assigning splits to reader {}", subtaskId);
+ enumeratorContext.assignSplit(cassandraSplit, subtaskId);
+ } else {
+ LOG.info(
+ "No split assigned to reader {} because the enumerator has
no unassigned split left",
+ subtaskId);
+ }
+ if (!state.hasMoreSplits()) {
+ LOG.info(
+ "No more CassandraSplits to assign. Sending
NoMoreSplitsEvent to reader {}.",
+ subtaskId);
+ enumeratorContext.signalNoMoreSplits(subtaskId);
+ }
Review Comment:
sketchy thing, not ready for review. But yes you're right
--
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.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
