Add support for XGROUP CREATE … MKSTREAM

[wenerme]

Bug Report

I can use MKSTREAM option in cli

127.0.0.1:6379> XGROUP CREATE s g1 $ MKSTREAM
OK

But can not use MKSTREAM with lettuce client, lettuce xgroupCreate don't support option, so I create
CommandArgs by myself, but the result is not same.

        RedisClient client = RedisClient.create("redis://localhost");
        StatefulRedisConnection<String, String> connection = client.connect();
        String streamName = "s";
        String groupName = "g1";

        StringCodec codec = StringCodec.UTF8;
        CommandArgs<String, String> args = new CommandArgs<>(codec)
                .add(CommandType.XGROUP)
                .add(CommandKeyword.CREATE)
                .add(streamName)
                .add(groupName)
                .add("$")
                .add("MKSTREAM");
        System.out.println(args.toCommandString());
        String status = connection.sync().dispatch(CommandType.XGROUP, new StatusOutput<>(codec), args);
        System.out.println(status);

Current Behavior

io.lettuce.core.RedisCommandExecutionException: ERR The XGROUP subcommand requires the key to exist. Note that for CREATE you may want to use the MKSTREAM option to create an empty stream automatically.

	at io.lettuce.core.ExceptionFactory.createExecutionException(ExceptionFactory.java:135)
	at io.lettuce.core.LettuceFutures.awaitOrCancel(LettuceFutures.java:122)
	at io.lettuce.core.FutureSyncInvocationHandler.handleInvocation(FutureSyncInvocationHandler.java:69)
	at io.lettuce.core.internal.AbstractInvocationHandler.invoke(AbstractInvocationHandler.java:80)
	at com.sun.proxy.$Proxy4.dispatch(Unknown Source)
	at test.io.RedisPlay.testThis(RedisPlay.java:55)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
	at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
	at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
	at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
	at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
	at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
	at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)
	at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
	at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)
Caused by: io.lettuce.core.RedisCommandExecutionException: ERR The XGROUP subcommand requires the key to exist. Note that for CREATE you may want to use the MKSTREAM option to create an empty stream automatically.
	at io.lettuce.core.ExceptionFactory.createExecutionException(ExceptionFactory.java:135)
	at io.lettuce.core.ExceptionFactory.createExecutionException(ExceptionFactory.java:108)
	at io.lettuce.core.protocol.AsyncCommand.completeResult(AsyncCommand.java:120)
	at io.lettuce.core.protocol.AsyncCommand.complete(AsyncCommand.java:111)
	at io.lettuce.core.protocol.CommandWrapper.complete(CommandWrapper.java:59)
	at io.lettuce.core.protocol.CommandHandler.complete(CommandHandler.java:646)
	at io.lettuce.core.protocol.CommandHandler.decode(CommandHandler.java:604)
	at io.lettuce.core.protocol.CommandHandler.channelRead(CommandHandler.java:556)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
	at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
	at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1414)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:945)
	at io.netty.channel.kqueue.AbstractKQueueStreamChannel$KQueueStreamUnsafe.readReady(AbstractKQueueStreamChannel.java:551)
	at io.netty.channel.kqueue.AbstractKQueueChannel$AbstractKQueueUnsafe.readReady(AbstractKQueueChannel.java:396)
	at io.netty.channel.kqueue.KQueueEventLoop.processReady(KQueueEventLoop.java:195)
	at io.netty.channel.kqueue.KQueueEventLoop.run(KQueueEventLoop.java:269)
	at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:886)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.lang.Thread.run(Thread.java:748)

Input Code

• Java/Kotlin/Scala/Groovy/… or Repo link if applicable:

// your code here;

Expected behavior/code

A clear and concise description of what you expected to happen (or code).

Environment

        <dependency>
            <groupId>io.lettuce</groupId>
            <artifactId>lettuce-core</artifactId>
            <version>5.1.1.RELEASE</version>
        </dependency>

127.0.0.1:6379> info server
# Server
redis_version:5.0.0
redis_git_sha1:00000000
redis_git_dirty:0
redis_build_id:2693b64502c71820

Possible Solution

Additional context

[mp911de]

Labeling this ticket as enhancement since the Redis API changed.

[mp911de]

Please note that the code above contains a bug. You're already specifying CommandType.XGROUP as command type, so it must be removed from CommandArgs. The command args should look like:

CommandArgs<String, String> args = new CommandArgs<>(codec)
        .add(CommandKeyword.CREATE)
        .add(streamName)
        .add(groupName)
        .add("$")
        .add("MKSTREAM");