def METHOD_NAME(ctx, destination, border, output_type, allow_dangerous_operation):
    if not allow_dangerous_operation:
        click.secho("DON'T RUN ON PRODUCTION MAY CAUSE OVERSIZED KAFKA MESSAGE",
                    blink=True, bold=True)
        return
    message = create_dump_state(ctx.correlation_id, destination=destination)
    LOG.debug('command = {}'.format(message.serialize()))
    with receive_with_context_async(ctx) as records:
        send_with_context(ctx, message.serialize())
    if output_type == 'table':
        print_table(records, border)
    elif output_type == 'json':
        for record in records:
            data = json.loads(record.value)
            print(pprint.pformat(data))