+ Mono<RecordMetadata> sendShardAssignedEvent(int shard)
+ {
+ EventShardAssigned to = EventShardAssigned.of(shard, instanceUri);
+
+ Sinks.One sink = Sinks.unsafe().one();
+
+ ProducerRecord<String, AbstractMessageTo> record =
+ new ProducerRecord<>(
+ topic,
+ Integer.toString(shard),
+ to);
+
+ producer.send(record, ((metadata, exception) ->
+ {
+ if (metadata != null)
+ {
+ log.info("Successfully sent shard assigned event for shard: {}", shard);
+ sink.emitValue(metadata, Sinks.EmitFailureHandler.FAIL_FAST);
+ }
+ else
+ {
+ // On send-failure
+ log.error(
+ "Could not send shard assigned event for shard {}: {}",
+ shard,
+ exception);
+ sink.emitError(exception, Sinks.EmitFailureHandler.FAIL_FAST);
+ }
+ }));
+
+ return sink.asMono();
+ }
+
+ Mono<RecordMetadata> sendShardRevokedEvent(int shard)
+ {
+ EventShardRevoked to = EventShardRevoked.of(shard, instanceUri);
+
+ Sinks.One sink = Sinks.unsafe().one();
+
+ ProducerRecord<String, AbstractMessageTo> record =
+ new ProducerRecord<>(
+ topic,
+ Integer.toString(shard),
+ to);
+
+ producer.send(record, ((metadata, exception) ->
+ {
+ if (metadata != null)
+ {
+ log.info("Successfully sent shard revoked event for shard: {}", shard);
+ sink.emitValue(metadata, Sinks.EmitFailureHandler.FAIL_FAST);
+ }
+ else
+ {
+ // On send-failure
+ log.error(
+ "Could not send shard revoked event for shard {}: {}",
+ shard,
+ exception);
+ sink.emitError(exception, Sinks.EmitFailureHandler.FAIL_FAST);
+ }
+ }));
+
+ return sink.asMono();
+ }
+