How to load balance leader using zookeeper and spring integration

Question

Using spring integration and zookeeper, one can implement a leader to perform activities such as polling.

However how do we distribute the leader responsibility to all nodes in the cluster to load balance?

Given below code, once the application starts, I see that the same node is maintaining the leader role and fetching events. I want to distribute this activity to every node in the cluster to better load balance.

Is there any way I can schedule each node in the cluster to gain leadership and revoke in round robin manner?

    @Bean
    public LeaderInitiatorFactoryBean fooLeaderInitiator(CuratorFramework client) {
        new LeaderInitiatorFactoryBean()
                .setClient(client)
                .setPath("/foofeed")
                .setRole("foo");
    }

    @Bean
    @InboundChannelAdapter(channel = "fooIncomingEvents", autoStartup = "false", poller = @Poller(fixedDelay = "5000"))
    @Role("foo")
    public FooTriggerMessageSource fooInboundChannelAdapter() {
        new FooMessageSource("foo")
    }

Answer 1

I could simulate load balancing using below code. Not sure if this is the correct approach. I could see fetching events log statement only from one node at a time in the cluster. This code yields leadership after performing gaining leadership and performing its job.

@Bean
public LeaderInitiator fooLeaderInitiator(CuratorFramework client, 
        FooPollingCandidate fooPollingCandidate) {
    LeaderInitiator leader = new LeaderInitiator(client, fooPollingCandidate, zooKeeperNamespace)
    leader.start()
    leader
}

@Component
class FooPollingCandidate extends DefaultCandidate {
    final Logger log = LoggerFactory.getLogger(this.getClass());

    FooPollingCandidate() {
        super("fooPoller", "foo")
    }

    @Override
    void onGranted(Context ctx) {
        log.debug("Leadership granted {}", ctx)
        pullEvents()
        ctx.yield();
    }

    @Override
    void onRevoked(Context ctx) {
        log.debug("Leadership revoked")
    }

    @Override
    void yieldLeadership() {
        log.debug("yielding Leadership")
    }

    //pull events and drop them on any channel needed
    void pullEvents() {
        log.debug("fetching events")
        //simulate delay
        sleep(5000)
    }
}

Answer 2

What you are suggesting is an abuse of the leader election technology, which is intended for warm failover when the current leader fails, manually yielding leadership after each event is an anti-pattern

What you probably want is competing pollers where all pollers are active, but use a shared store to prevent duplicate processing.

For example, if you are polling a shared directory for files to process, you would use a FileSystemPersistentFileListFilter with a shared MetadataStore (such as the zookeeper implementation) to prevent multiple instances from processing the same file.

You can use the same technique (shared metadata store) for any polled message source.