[DISCUSSION] Workload splitting through Hazelcast

Start by reviewing the proposed HazelcastPartitionFilter and the adaptations to CloudDiscoveryService, the triggers, InstanceRegistry, and InstanceDiscoveryListener in the issue. Check how these components currently handle discovery and events, then assess whether the proposal covers all relevant code paths and preserves expected behavior across nodes. Done means reaching agreement on an approach before any large implementation PR.

Hello guys.

Recently I've been thinking to ways to make the current (half) clustering work fully so that jobs (e.g. StatusUpdateTrigger, InfoUpdateTrigger, NotificationTrigger and so on) are executed only on one one node while the whole state is anyway kept in sync with Hazelcast-specific collections.

I've therefore been working a bit with AI on this and this was a suggestion I got from it which, when I see it, code quality aside, it kind of makes sense to me:

    @Component
    public static class HazelcastPartitionFilter {
        private final HazelcastInstance hazelcastInstance;

        public HazelcastPartitionFilter(HazelcastInstance hazelcastInstance) {
            this.hazelcastInstance = hazelcastInstance;
        }

        public boolean isMyPartition(InstanceId instanceId) {
            if (instanceId == null || instanceId.getValue() == null) {
                return false;
            }
            var partitionService = hazelcastInstance.getPartitionService();
            Partition partition = partitionService.getPartition(instanceId.getValue());
            return partition.getOwner() != null && partition.getOwner().localMember();
        }
    }

    public static class HazelcastShardedDiscoveryService extends CloudDiscoveryService {
        private final DiscoveryClient discoveryClient;
        private final HazelcastPartitionFilter partitionFilter;

        public HazelcastShardedDiscoveryService(DiscoveryClient discoveryClient, 
                                                HazelcastPartitionFilter partitionFilter) {
            super(discoveryClient);
            this.discoveryClient = discoveryClient;
            this.partitionFilter = partitionFilter;
        }

        @Override
        public Flux<Registration> convertToRegistrations(String serviceId) {
            return Flux.fromIterable(discoveryClient.getInstances(serviceId))
                    .map(instance -> this.convertToRegistration(serviceId, instance))
                    .filter(reg -> partitionFilter.isMyPartition(reg.getInstanceId()));
        }
    }

    @Bean
    @Primary
    public CloudDiscoveryService cloudDiscoveryService(DiscoveryClient discoveryClient, 
                                                       HazelcastPartitionFilter partitionFilter) {
        return new HazelcastShardedDiscoveryService(discoveryClient, partitionFilter);
    }

    @Bean
    @Primary
    public StatusUpdateTrigger statusUpdateTrigger(StatusUpdater statusUpdater, HazelcastPartitionFilter partitionFilter) {
        return new StatusUpdateTrigger(statusUpdater) {
            @Override
            protected Publisher<Void> handle(InstanceEvent event) {
                if (partitionFilter.isMyPartition(event.getInstance())) {
                    return super.handle(event);
                }
                return Mono.empty();
            }
        };
    }

    @Bean
    @Primary
    public InfoUpdateTrigger infoUpdateTrigger(InfoUpdater infoUpdater, HazelcastPartitionFilter partitionFilter) {
        return new InfoUpdateTrigger(infoUpdater) {
            @Override
            protected Publisher<Void> handle(InstanceEvent event) {
                if (partitionFilter.isMyPartition(event.getInstance())) {
                    return super.handle(event);
                }
                return Mono.empty();
            }
        };
    }

    @Bean
    @Primary
    public EndpointDetectionTrigger endpointDetectionTrigger(EndpointDetector endpointDetector, HazelcastPartitionFilter partitionFilter) {
        return new EndpointDetectionTrigger(endpointDetector) {
            @Override
            protected Publisher<Void> handle(InstanceEvent event) {
                if (partitionFilter.isMyPartition(event.getInstance())) {
                    return super.handle(event);
                }
                return Mono.empty();
            }
        };
    }

    @Bean
    @Primary
    public NotificationTrigger notificationTrigger(Notifier notifier, 
                                                   InstanceRepository repository, 
                                                   HazelcastPartitionFilter partitionFilter) {
        return new NotificationTrigger(notifier, repository) {
            @Override
            protected Publisher<Void> handle(InstanceEvent event) {
                if (partitionFilter.isMyPartition(event.getInstance())) {
                    return super.handle(event);
                }
                return Mono.empty();
            }
        };
    }

    @Bean
    @Primary
    public InstanceRegistry instanceRegistry(InstanceRepository repository,
                                             InstanceIdGenerator idGenerator,
                                             HazelcastPartitionFilter partitionFilter) {
        return new InstanceRegistry(repository, idGenerator) {
            @Override
            public Mono<InstanceId> register(Registration registration) {
                if (partitionFilter.isMyPartition(idGenerator.generateId(registration))) {
                    return super.register(registration);
                }
                return Mono.empty();
            }
        };
    }

    @Bean
    @Primary
    public InstanceDiscoveryListener instanceDiscoveryListener(CloudDiscoveryService discoveryService,
                                                               InstanceRegistry registry,
                                                               InstanceRepository repository) {
        return new InstanceDiscoveryListener(discoveryService, registry, repository) {
            @Override
            public void discover() {
                // Force the sync
                super.discover();
            }
        };
    }

I wanted to open this ticket to discuss with you the solution above and, in particular:

• if there any other code paths which I'm not aware of and that would require a similar adaptation
• if you think the solution above makes sense and it could actually work
• if you're willing to accept a PR with a polished version of the above (I can already anticipate it will be big)