Super stream creation using Java Client API

[vinodreddyb]

Is your feature request related to a problem? Please describe.

As per documentation super stream can be created using CLI

rabbitmq-streams add_super_stream invoices --partitions 3

But how do I create same with Java Client library. I don't find any example in documentation.

Describe the solution you'd like

Provide any example to create super stream from java API

Describe alternatives you've considered

No response

Additional context

No response

[lukebakken]

Hello, thanks for using RabbitMQ.

Just so you know, if a GitHub repository has both "Issues" and "Discussions", you should use "Discussions" for questions. Issues are for actual bugs.

[acogoluegnes]

It's not possible with the stream protocol, we may introduce this feature later though. In the meantime, it is still possible to use plain AMQP, here is how it is done with the AMQP Java client in the test suite:

rabbitmq-stream-java-client/src/test/java/com/rabbitmq/stream/impl/TestUtils.java

Lines 284 to 321 in 4fa7e81

	static void declareSuperStreamTopology(Connection connection, String superStream, int partitions)
	throws Exception {
	declareSuperStreamTopology(
	connection,
	superStream,
	IntStream.range(0, partitions).mapToObj(String::valueOf).toArray(String[]::new));
	}
	static void declareSuperStreamTopology(Connection connection, String superStream, String... rks)
	throws Exception {
	try (Channel ch = connection.createChannel()) {
	ch.exchangeDeclare(
	superStream,
	BuiltinExchangeType.DIRECT,
	true,
	false,
	Collections.singletonMap("x-super-stream", true));
	List<Tuple2<String, Integer>> bindings = new ArrayList<>(rks.length);
	for (int i = 0; i < rks.length; i++) {
	bindings.add(Tuple.of(rks[i], i));
	}
	// shuffle the order to make sure we get in the correct order from the server
	Collections.shuffle(bindings);
	for (Tuple2<String, Integer> binding : bindings) {
	String routingKey = binding._1();
	String partitionName = superStream + "-" + routingKey;
	ch.queueDeclare(
	partitionName, true, false, false, Collections.singletonMap("x-queue-type", "stream"));
	ch.queueBind(
	partitionName,
	superStream,
	routingKey,
	Collections.singletonMap("x-stream-partition-order", binding._2()));
	}
	}
	}