imaffe commented on code in PR #20725:
URL: https://github.com/apache/flink/pull/20725#discussion_r960272872
##########
flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SharedSplitAssigner.java:
##########
@@ -83,66 +75,8 @@ public List<TopicPartition>
registerTopicPartitions(Set<TopicPartition> fetchedP
@Override
public void addSplitsBack(List<PulsarPartitionSplit> splits, int
subtaskId) {
- Set<PulsarPartitionSplit> pendingPartitionSplits =
- sharedPendingPartitionSplits.computeIfAbsent(subtaskId, id ->
new HashSet<>());
- pendingPartitionSplits.addAll(splits);
- }
-
- @Override
- public Optional<SplitsAssignment<PulsarPartitionSplit>> createAssignment(
- List<Integer> readers) {
- if (readers.isEmpty()) {
- return Optional.empty();
- }
-
- Map<Integer, List<PulsarPartitionSplit>> assignMap = new HashMap<>();
- for (Integer reader : readers) {
- Set<PulsarPartitionSplit> pendingSplits =
sharedPendingPartitionSplits.remove(reader);
- if (pendingSplits == null) {
- pendingSplits = new HashSet<>();
- }
-
- Set<String> assignedSplits =
- readerAssignedSplits.computeIfAbsent(reader, r -> new
HashSet<>());
-
- for (TopicPartition partition : appendedPartitions) {
- String partitionName = partition.toString();
- if (!assignedSplits.contains(partitionName)) {
- pendingSplits.add(new PulsarPartitionSplit(partition,
stopCursor));
- assignedSplits.add(partitionName);
- }
- }
-
- if (!pendingSplits.isEmpty()) {
- assignMap.put(reader, new ArrayList<>(pendingSplits));
- }
- }
-
- if (assignMap.isEmpty()) {
- return Optional.empty();
- } else {
- return Optional.of(new SplitsAssignment<>(assignMap));
- }
- }
-
- @Override
- public boolean noMoreSplits(Integer reader) {
- Set<PulsarPartitionSplit> pendingSplits =
sharedPendingPartitionSplits.get(reader);
- Set<String> assignedSplits = readerAssignedSplits.get(reader);
-
- return !enablePartitionDiscovery
- && initialized
- && (pendingSplits == null || pendingSplits.isEmpty())
- && (assignedSplits != null && assignedSplits.size() ==
appendedPartitions.size());
- }
-
- @Override
- public PulsarSourceEnumState snapshotState() {
- return new PulsarSourceEnumState(
- appendedPartitions,
- new HashSet<>(),
- sharedPendingPartitionSplits,
- readerAssignedSplits,
- initialized);
+ Set<PulsarPartitionSplit> pendingSplits =
+ this.pendingPartitionSplits.computeIfAbsent(subtaskId, id ->
new HashSet<>());
Review Comment:
can we reuse the addSplitToPendingList() ? It is easier to track when the
pendingList is changed if all changes to pendingList uses the same method
--
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
