echauchot commented on code in PR #3:
URL:
https://github.com/apache/flink-connector-cassandra/pull/3#discussion_r1100131432
##########
flink-connector-cassandra/src/main/java/org/apache/flink/connector/cassandra/source/enumerator/CassandraSplitEnumerator.java:
##########
@@ -115,29 +109,22 @@ private List<CassandraSplit> discoverSplits() {
@Override
public void addSplitsBack(List<CassandraSplit> splits, int subtaskId) {
LOG.info("Add {} splits back to CassandraSplitEnumerator.",
splits.size());
- state.addNewSplits(splits, enumeratorContext.currentParallelism());
- assignUnprocessedSplitsToReader(subtaskId);
+ state.addNewSplits(splits);
}
@Override
public void addReader(int subtaskId) {
LOG.info("Adding reader {} to CassandraSplitEnumerator.", subtaskId);
- assignUnprocessedSplitsToReader(subtaskId);
+ assignUnprocessedSplitToReader(subtaskId);
}
- private void assignUnprocessedSplitsToReader(int readerId) {
+ private void assignUnprocessedSplitToReader(int readerId) {
checkReaderRegistered(readerId);
-
- final Set<CassandraSplit> splitsForReader =
state.getSplitsForReader(readerId);
- if (splitsForReader != null && !splitsForReader.isEmpty()) {
- Map<Integer, List<CassandraSplit>> assignment = new HashMap<>();
- assignment.put(readerId, Lists.newArrayList(splitsForReader));
- LOG.info("Assigning splits to reader {}", assignment);
- enumeratorContext.assignSplits(new SplitsAssignment<>(assignment));
- }
-
- // periodically partition discovery is disabled, signal
NoMoreSplitsEvent to the reader
- LOG.debug(
+ final CassandraSplit cassandraSplit = state.getASplit();
Review Comment:
:+1:
--
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
