- return new ShardingPublisherStrategy() {
- @Override
- public Mono<String> publishOwnership(int shard)
- {
- return Mono.just(Integer.toString(shard));
- }
- };
+ String[] parts = properties.getKafka().getHaproxyRuntimeApi().split(":");
+ InetSocketAddress haproxyAddress = new InetSocketAddress(parts[0], Integer.valueOf(parts[1]));
+ return new HaproxyShardingPublisherStrategy(
+ haproxyAddress,
+ properties.getKafka().getHaproxyMap(),
+ properties.getInstanceId());