code: def test_stateful_index_search_container(
      shards, tmpdir, stateful_exec_docker_image_built
  ):
      replicas = 3
      peer_ports = {}
      for shard in range(shards):
          peer_ports[shard] = [random_port() for _ in range(replicas)]
  
      dep = Deployment(
          uses="docker://stateful-exec",
          replicas=replicas,
          stateful=True,
          raft_configuration={
              "snapshot_interval": 10,
              "snapshot_threshold": 5,
              "trailing_logs": 10,
              "LogLevel": "INFO",
          },
          shards=shards,
          workspace="/workspace/tmp",
          volumes=[str(tmpdir) + ":" + "/workspace/tmp"],
          peer_ports=peer_ports,
          polling={
              "/index": "ANY",
              "/search": "ALL",
              "/similarity": "ALL",
          },
      )
      with dep:
          index_da = DocumentArray[TextDocWithId](
              [
                  TextDocWithId(id=f"{i}", text=f"ID {i}")
                  for i in range(100)
              ]
          )
          search_da = DocumentArray[TextDocWithId](
              [TextDocWithId(id=f"{i}") for i in range(100)]
          )
          dep.index(
              inputs=index_da,
              request_size=1,
              return_type=DocumentArray[TextDocWithId],
          )
  
          # allowing some time for the state to be replicated
          time.sleep(20)
          # checking against the main read replica
          assert_is_indexed(dep, search_da)
          assert_all_replicas_indexed(dep, search_da, key="num")
          time.sleep(10)
  
      dep_restore = Deployment(
          uses="docker://stateful-exec",
          replicas=replicas,
          stateful=True,
          raft_configuration={
              "snapshot_interval": 10,
              "snapshot_threshold": 5,
              "trailing_logs": 10,
              "LogLevel": "INFO",
          },
          shards=shards,
          workspace="/workspace/tmp",
          volumes=[str(tmpdir) + ":" + "/workspace/tmp"],
          peer_ports=peer_ports,
          polling={
              "/index": "ANY",
              "/search": "ALL",
              "/similarity": "ALL",
          },
      )
      # test restoring
      with dep_restore:
          index_da = DocumentArray[TextDocWithId](
              [
                  TextDocWithId(id=f"{i}", text=f"ID {i}")
                  for i in range(100, 200)
              ]
          )
          dep_restore.index(
              inputs=index_da,
              request_size=1,
              return_type=DocumentArray[TextDocWithId],
          )
          time.sleep(20)
          search_da = DocumentArray[TextDocWithId](
              [TextDocWithId(id=f"{i}") for i in range(200)]
          )
          assert_all_replicas_indexed(dep_restore, search_da, key="num")
          time.sleep(10)
  
file_path: jina/tests/integration/stateful/test_stateful.py
imports: import time
  import pytest
  import os
  from jina import Client, Document, DocumentArray, Flow, Deployment
  from typing import Dict, List
  from jina.helper import random_port
  from tests.integration.stateful.stateful_no_snapshot_exec.executor import MyStateExecutorNoSnapshot
  from tests.integration.stateful.stateful_snapshot_exec.executor import MyStateExecutor
  from jina._docarray import docarray_v2
label_desc: Use image node_id 7 for calling a global function with example usage: test_stateful_index_search_container(shards, tmpdir, stateful_exec_docker_image_built) without return types