abstract void execute(Admin admin, Namespace ns, PrintStream out) throws Exception;
@Test
    public void testNewBrokerAbortTransaction() throws Exception {
        TopicPartition topicPartition = new TopicPartition("foo", 5);
        long startOffset = 9173;
        long producerId = 12345L;
        short producerEpoch = 15;
        int coordinatorEpoch = 76;

        String[] args = new String[] {
            "--bootstrap-server",
            "localhost:9092",
            "abort",
            "--topic",
            topicPartition.topic(),
            "--partition",
            String.valueOf(topicPartition.partition()),
            "--start-offset",
            String.valueOf(startOffset)
        };

        DescribeProducersResult describeResult = Mockito.mock(DescribeProducersResult.class);
        KafkaFuture<PartitionProducerState> describeFuture = completedFuture(
            new PartitionProducerState(singletonList(
                new ProducerState(producerId, producerEpoch, 1300, 1599509565L,
                    OptionalInt.of(coordinatorEpoch), OptionalLong.of(startOffset))
            )));

        AbortTransactionResult abortTransactionResult = Mockito.mock(AbortTransactionResult.class);
        KafkaFuture<Void> abortFuture = completedFuture(null);
        AbortTransactionSpec expectedAbortSpec = new AbortTransactionSpec(
            topicPartition, producerId, producerEpoch, coordinatorEpoch);

        Mockito.when(describeResult.partitionResult(topicPartition)).thenReturn(describeFuture);
        Mockito.when(admin.describeProducers(singleton(topicPartition))).thenReturn(describeResult);

        Mockito.when(abortTransactionResult.all()).thenReturn(abortFuture);
        Mockito.when(admin.abortTransaction(expectedAbortSpec)).thenReturn(abortTransactionResult);

        execute(args);
        assertNormalExit();
    }