code: def test_kinesis_stream(self):
      import random
  
      kinesisAppName = "KinesisStreamTests-%d" % abs(
          random.randint(0, 10000000)
      )
      kinesisTestUtils = self.ssc._jvm.org.apache.spark.streaming.kinesis.KinesisTestUtils(
          2
      )
      try:
          kinesisTestUtils.createStream()
          aWSCredentials = kinesisTestUtils.getAWSCredentials()
          stream = KinesisUtils.createStream(
              self.ssc,
              kinesisAppName,
              kinesisTestUtils.streamName(),
              kinesisTestUtils.endpointUrl(),
              kinesisTestUtils.regionName(),
              InitialPositionInStream.LATEST,
              10,
              MetricsLevel.DETAILED,
              StorageLevel.MEMORY_ONLY,
              aWSCredentials.getAWSAccessKeyId(),
              aWSCredentials.getAWSSecretKey(),
          )
  
          outputBuffer = []
  
          def get_output(_, rdd):
              for e in rdd.collect():
                  outputBuffer.append(e)
  
          stream.foreachRDD(get_output)
          self.ssc.start()
  
          testData = [i for i in range(1, 11)]
          expectedOutput = set([str(i) for i in testData])
          start_time = time.time()
          while time.time() - start_time < 120:
              kinesisTestUtils.pushData(testData)
              if expectedOutput == set(outputBuffer):
                  break
              time.sleep(10)
          self.assertEqual(expectedOutput, set(outputBuffer))
      except BaseException:
          import traceback
  
          traceback.print_exc()
          raise
      finally:
          self.ssc.stop(False)
          kinesisTestUtils.deleteStream()
          kinesisTestUtils.deleteDynamoDBTable(kinesisAppName)
  
file_path: spark/python/pyspark/streaming/tests/test_kinesis.py
imports: import time
  import unittest
  from pyspark import StorageLevel
  from pyspark.streaming.kinesis import KinesisUtils, InitialPositionInStream, MetricsLevel
  from pyspark.testing.streamingutils import should_test_kinesis, kinesis_requirement_message, PySparkStreamingTestCase
label_desc: Use image node_id 2 for calling the KinesisStreamTests obj's underlying member method code with example usage: obj.test_kinesis_stream() without return types