org.apache.flink.connector.pulsar.source.enumerator.assigner

Class KeySharedSplitAssigner

java.lang.Object

org.apache.flink.connector.pulsar.source.enumerator.assigner.KeySharedSplitAssigner

All Implemented Interfaces:

SplitAssigner

@Internal
public class KeySharedSplitAssigner
extends Object

This assigner is used for SubscriptionType.Key_Shared subscription.

Field Summary

Fields

Modifier and Type	Field and Description
protected Set<TopicPartition>	appendedPartitions
protected SplitEnumeratorContext<PulsarPartitionSplit>	context
protected boolean	enablePartitionDiscovery
protected boolean	initialized
protected Map<Integer,Set<PulsarPartitionSplit>>	pendingPartitionSplits
protected StopCursor	stopCursor

Constructor Summary

Constructors

Constructor and Description
KeySharedSplitAssigner(StopCursor stopCursor, boolean enablePartitionDiscovery, SplitEnumeratorContext<PulsarPartitionSplit> context, PulsarSourceEnumState enumState)

Method Summary

Modifier and Type	Method and Description
void	addSplitsBack(List<PulsarPartitionSplit> splits, int subtaskId) Add a split back to the split assigner if the reader fails.
protected void	addSplitToPendingList(int readerId, PulsarPartitionSplit split) Add split to pending lists.
Optional<SplitsAssignment<PulsarPartitionSplit>>	createAssignment(List<Integer> readers) Create a split assignment from the current readers.
boolean	noMoreSplits(Integer reader) It would return true only if periodically partition discovery is turned off, the initializing partition discovery has finished, AND there are no pending splits for assignment.
protected int	partitionOwner(TopicPartition partition) Returns the index of the target subtask that a specific partition should be assigned to.
List<TopicPartition>	registerTopicPartitions(Set<TopicPartition> fetchedPartitions) Add the current available partitions into assigner.
PulsarSourceEnumState	snapshotState() Snapshot the current assign state into checkpoint.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

stopCursor

protected final StopCursor stopCursor

enablePartitionDiscovery

protected final boolean enablePartitionDiscovery

context

protected final SplitEnumeratorContext<PulsarPartitionSplit> context

appendedPartitions

protected final Set<TopicPartition> appendedPartitions

pendingPartitionSplits

protected final Map<Integer,Set<PulsarPartitionSplit>> pendingPartitionSplits

initialized

protected boolean initialized

Constructor Detail

KeySharedSplitAssigner

public KeySharedSplitAssigner(StopCursor stopCursor,
                              boolean enablePartitionDiscovery,
                              SplitEnumeratorContext<PulsarPartitionSplit> context,
                              PulsarSourceEnumState enumState)

Method Detail

registerTopicPartitions

public List<TopicPartition> registerTopicPartitions(Set<TopicPartition> fetchedPartitions)

Description copied from interface: SplitAssigner

Add the current available partitions into assigner.

Parameters:

fetchedPartitions - The available partitions queried from Pulsar broker.

Returns:

New topic partitions compare to previous registered partitions.

addSplitsBack

public void addSplitsBack(List<PulsarPartitionSplit> splits,
                          int subtaskId)

Description copied from interface: SplitAssigner

Add a split back to the split assigner if the reader fails. We would try to reassign the split or add it to the pending list.

createAssignment

public Optional<SplitsAssignment<PulsarPartitionSplit>> createAssignment(List<Integer> readers)

Description copied from interface: SplitAssigner

Create a split assignment from the current readers.

Specified by:

createAssignment in interface SplitAssigner

noMoreSplits

public boolean noMoreSplits(Integer reader)

Description copied from interface: SplitAssigner

It would return true only if periodically partition discovery is turned off, the initializing partition discovery has finished, AND there are no pending splits for assignment.

Specified by:

noMoreSplits in interface SplitAssigner

snapshotState

public PulsarSourceEnumState snapshotState()

Description copied from interface: SplitAssigner

Snapshot the current assign state into checkpoint.

Specified by:

snapshotState in interface SplitAssigner

addSplitToPendingList

protected void addSplitToPendingList(int readerId,
                                     PulsarPartitionSplit split)

Add split to pending lists.

partitionOwner

protected int partitionOwner(TopicPartition partition)

Returns the index of the target subtask that a specific partition should be assigned to. It's inspired by the KafkaSourceEnumerator.getSplitOwner()

The resulting distribution of partition has the following contract:

• 1. Uniformly distributed across subtasks.
• 2. Partitions are round-robin distributed (strictly clockwise w.r.t. ascending subtask indices) by using the partition id as the offset from a starting index (i.e., the index of the subtask which partition 0 of the topic will be assigned to, determined using the topic name).

Parameters:

partition - The Pulsar partition to assign.

Returns:

The id of the reader that owns this partition.