### Basic tests

* Bring up a 3-node Kafka cluster
    * Configured to have 10+ partitions per topic

    volumes:
      - type: bind
        source: ./strimzi-ca-certs
        target: /opt/kafka/cluster-ca-certs
        read_only: true
      - type: bind
        source: ./strimzi-broker-certs
        target: /opt/kafka/broker-certs
        read_only: true
      - type: bind
        source: ./strimzi-client-ca
        target: /opt/kafka/client-ca-certs
        read_only: true
      - type: bind
        source: ./strimzi-listener-certs
        target: /opt/kafka/certificates/custom-mtls-9094-certs
        read_only: true
      - type: bind
        source: ./strimzi-listener-certs
        target: /opt/kafka/certificates/custom-oauth-9093-certs
        read_only: true
      - type: bind
        source: ./strimzi-listener-certs
        target: /opt/kafka/certificates/oauth-oauth-9093-certs
        read_only: true
      - type: bind
        source: ./strimzi-kafka1-config
        target: /opt/kafka/custom-config
        read_only: true

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: messagebus
spec:
  type: pubsub.kafka
  version: v1
  ......
  metadata:
  - name: initialOffset
    value: oldest
  - name: backOffDuration
    value: 50ms

  metadata:
  - name: brokers
    value: localhost:19092,localhost:29092,localhost:39092
  - name: consumerGroup
    value: kafkaCertification1
  - name: authType
    value: "none"

  metadata:
  - name: brokers
    value: localhost:19094,localhost:29094,localhost:39094
  - name: consumerGroup
    value: kafkaCertification2
  - name: authType
    value: mtls
  - name: caCert
    value: |
      -----BEGIN CERTIFICATE-----      
			......
      -----END CERTIFICATE-----
  - name: clientCert
    value: |
      -----BEGIN CERTIFICATE-----      
			......
      -----END CERTIFICATE-----
  - name: clientKey
    value: |
      -----BEGIN RSA PRIVATE KEY-----
      ......
      -----END RSA PRIVATE KEY-----      

  metadata:
  - name: brokers
    value: localhost:19093,localhost:29093,localhost:39093
  - name: consumerGroup
    value: kafkaCertification2
  - name: authType
    value: "oidc"
  - name: oidcTokenEndpoint
    value: https://localhost:4443/oauth2/token
  - name: oidcClientID
    value: "dapr"
  - name: oidcClientSecret
    value: "dapr-test"
  - name: oidcScopes
    value: openid,kafka
  - name: caCert
    value: |
      -----BEGIN CERTIFICATE-----      
			......
      -----END CERTIFICATE-----

	log := logger.NewLogger("dapr.components")
	component := pubsub_loader.New("kafka", func() pubsub.PubSub {
		return pubsub_kafka.NewKafka(log)
	})

	// For Kafka, we should ensure messages are received in order.
	consumerGroup1 := watcher.NewOrdered()
	// This watcher is across multiple consumers in the same group
	// so exact ordering is not expected.
	consumerGroup2 := watcher.NewUnordered()

	// Application logic that tracks messages from a topic.
	application := func(messages *watcher.Watcher) app.SetupFn {
		return func(ctx flow.Context, s common.Service) error {
			// Simulate periodic errors.
			sim := simulate.PeriodicError(ctx, 100)

			// Setup the /orders event handler.
			return multierr.Combine(
				s.AddTopicEventHandler(&common.Subscription{
					PubsubName: "messagebus",
					Topic:      "neworder",
					Route:      "/orders",
				}, func(_ context.Context, e *common.TopicEvent) (retry bool, err error) {
					if err := sim(); err != nil {
						return true, err
					}

					// Track/Observe the data of the event.
					messages.Observe(e.Data)
					return false, nil
				}),
			)
		}
	}

const (
  numMessages       = 1000
  ）

// Test logic that sends messages to a topic and
	// verifies the application has received them.
	sendRecvTest := func(metadata map[string]string, messages ...*watcher.Watcher) flow.Runnable {
		_, hasKey := metadata[messageKey]
		return func(ctx flow.Context) error {
      // 获取client，里面封装还有dapr go sdk的 Client interface
			client := sidecar.GetClient(ctx, sidecarName1)

			// Declare what is expected BEFORE performing any steps
			// that will satisfy the test.
			msgs := make([]string, numMessages)
			for i := range msgs {
				msgs[i] = fmt.Sprintf("Hello, Messages %03d", i)
			}
			for _, m := range messages {
				m.ExpectStrings(msgs...)
			}
			// If no key it provided, create a random one.
			// For Kafka, this will spread messages across
			// the topic's partitions.
			if !hasKey {
				metadata[messageKey] = uuid.NewString()
			}

			// Send events that the application above will observe.
			ctx.Log("Sending messages!")
			for _, msg := range msgs {
				ctx.Logf("Sending: %q", msg)
        // 通过 dapr client 的 PublishEvent 方法发送消息到topic
        // 对应的 pubsubName 为 "messagebus"，topicName 为 "neworder"
        // 消息内容为前面准备的 "Hello, Messages %03d"
				err := client.PublishEvent(
					ctx, pubsubName, topicName, msg,
					dapr.PublishEventWithMetadata(metadata))
				require.NoError(ctx, err, "error publishing message")
			}

			// Do the messages we observed match what we expect?
			for _, m := range messages {
        // 一分钟内应该能收到全部消息
				m.Assert(ctx, time.Minute)
			}

			return nil
		}
	}

clusterName       = "kafkacertification"
dockerComposeYAML = "docker-compose.yml"		

// Run Kafka using Docker Compose.
Step(dockercompose.Run(clusterName, dockerComposeYAML)).

var (
	brokers          = []string{"localhost:19092", "localhost:29092", "localhost:39092"}
)
Step("wait for broker sockets",
			network.WaitForAddresses(5*time.Minute, brokers...)).

		Step("wait for kafka readiness", retry.Do(time.Second, 30, func(ctx flow.Context) error {
			config := sarama.NewConfig()
			config.ClientID = "test-consumer"
			config.Consumer.Return.Errors = true

			// Create new consumer
			client, err := sarama.NewConsumer(brokers, config)
			if err != nil {
				return err
			}
			defer client.Close()

			// Ensure the brokers are ready by attempting to consume
			// a topic partition.
			_, err = client.ConsumePartition("myTopic", 0, sarama.OffsetOldest)

			return err
		})).

		Step("wait for Dapr OAuth client", retry.Do(10*time.Second, 6, func(ctx flow.Context) error {
			httpClient := &http.Client{
				Transport: &http.Transport{
					TLSClientConfig: &tls.Config{
						InsecureSkipVerify: true, // test server certificate is not trusted.
					},
				},
			}

			resp, err := httpClient.Get(oauthClientQuery)
			if err != nil {
				return err
			}
			if resp.StatusCode != 200 {
				return fmt.Errorf("oauth client query for 'dapr' not successful")
			}
			return nil
		})).

const (
	appID1            = "app-1"
  appPort           = 8000
)
consumerGroup1 := watcher.NewOrdered()

// Run the application logic above.
Step(app.Run(appID1, fmt.Sprintf(":%d", appPort),
             application(consumerGroup1))).

const (
	sidecarName1      = "dapr-1"
)

Step(sidecar.Run(sidecarName1,
			embedded.WithComponentsPath("./components/consumer1"),
			embedded.WithAppProtocol(runtime.HTTPProtocol, appPort),
			embedded.WithDaprGRPCPort(runtime.DefaultDaprAPIGRPCPort),
			embedded.WithDaprHTTPPort(runtime.DefaultDaprHTTPPort),
			runtime.WithPubSubs(component))).

const (
	appID2            = "app-2"
)
		// Run the second application.
		Step(app.Run(appID2, fmt.Sprintf(":%d", appPort+portOffset),
			application(consumerGroup2))).

const (
	sidecarName2      = "dapr-2"
)
		// Run the Dapr sidecar with the Kafka component.
		Step(sidecar.Run(sidecarName2,
			embedded.WithComponentsPath("./components/mtls-consumer"),
			embedded.WithAppProtocol(runtime.HTTPProtocol, appPort+portOffset),
			embedded.WithDaprGRPCPort(runtime.DefaultDaprAPIGRPCPort+portOffset),
			embedded.WithDaprHTTPPort(runtime.DefaultDaprHTTPPort+portOffset),
			embedded.WithProfilePort(runtime.DefaultProfilePort+portOffset),
			runtime.WithPubSubs(component))).

const (
	appID3            = "app-3"
)
		// Run the third application.
		Step(app.Run(appID3, fmt.Sprintf(":%d", appPort+portOffset*2),
			application(consumerGroup2))).

const (
	sidecarName3      = "dapr-3"
)
		// Run the Dapr sidecar with the Kafka component.
		Step(sidecar.Run(sidecarName3,
			embedded.WithComponentsPath("./components/oauth-consumer"),
			embedded.WithAppProtocol(runtime.HTTPProtocol, appPort+portOffset*2),
			embedded.WithDaprGRPCPort(runtime.DefaultDaprAPIGRPCPort+portOffset*2),
			embedded.WithDaprHTTPPort(runtime.DefaultDaprHTTPPort+portOffset*2),
			embedded.WithProfilePort(runtime.DefaultProfilePort+portOffset*2),
			runtime.WithPubSubs(component))).

Step("reset", flow.Reset(consumerGroup2)).

const（	
messageKey        = "partitionKey"	
）
// Set the partition key on all messages so they
	// are written to the same partition.
	// This allows for checking of ordered messages.
	metadata := map[string]string{
		messageKey: "test",
	}

// Send messages using the same metadata/message key so we can expect
// in-order processing.
Step("send and wait", sendRecvTest(metadata, consumerGroup1, consumerGroup2)).

// 先重置 consumerGroup2
Step("reset", flow.Reset(consumerGroup2)).		
// Send messages with random keys to test message consumption
// across more than one consumer group and consumers per group.
Step("send and wait", sendRecvTest(map[string]string{}, consumerGroup2)).

// Gradually stop each broker.
// This tests the components ability to handle reconnections
// when brokers are shutdown cleanly.
// 启动异步步骤，在后台发送消息
StepAsync("steady flow of messages to publish", &task,
          sendMessagesInBackground(consumerGroup1, consumerGroup2)).
// 5秒钟之后关闭 kafka1 broker
Step("wait", flow.Sleep(5*time.Second)).
Step("stop broker 1", dockercompose.Stop(clusterName, dockerComposeYAML, "kafka1")).

// 再过5秒钟之后关闭 kafka2 broker
Step("wait", flow.Sleep(5*time.Second)).
// Errors will likely start occurring here since quorum is lost.
Step("stop broker 2", dockercompose.Stop(clusterName, dockerComposeYAML, "kafka2")).

// 再过10秒钟之后关闭 kafka3 broker
Step("wait", flow.Sleep(10*time.Second)).
// Errors will definitely occur here.
Step("stop broker 3", dockercompose.Stop(clusterName, dockerComposeYAML, "kafka3")).

// 等待30秒后，重新启动 kafka1 / kafka2 / kafka3
Step("wait", flow.Sleep(30*time.Second)).
Step("restart broker 3", dockercompose.Start(clusterName, dockerComposeYAML, "kafka3")).
Step("restart broker 2", dockercompose.Start(clusterName, dockerComposeYAML, "kafka2")).
Step("restart broker 1", dockercompose.Start(clusterName, dockerComposeYAML, "kafka1")).

// Component should recover at this point.
// 再等30秒，组件应该能恢复工作
Step("wait", flow.Sleep(30*time.Second)).
// 检查消息已经都接收到
Step("assert messages", assertMessages(consumerGroup1, consumerGroup2)).

		// Simulate a network interruption.
		// This tests the components ability to handle reconnections
		// when Dapr is disconnected abnormally.
		StepAsync("steady flow of messages to publish", &task,
			sendMessagesInBackground(consumerGroup1, consumerGroup2)).
		Step("wait", flow.Sleep(5*time.Second)).
		//
		// Errors will occurring here.
		Step("interrupt network",
			network.InterruptNetwork(30*time.Second, nil, nil, "19092", "29092", "39092")).
		//
		// Component should recover at this point.
		Step("wait", flow.Sleep(30*time.Second)).
		Step("assert messages", assertMessages(consumerGroup1, consumerGroup2)).

// Reset and test that all messages are received during a
// consumer rebalance.
Step("reset", flow.Reset(consumerGroup2)).
// 后台异步发送消息
StepAsync("steady flow of messages to publish", &task,
          sendMessagesInBackground(consumerGroup2)).
// 15秒之后停止 sidecar dapr-2
Step("wait", flow.Sleep(15*time.Second)).
Step("stop sidecar 2", sidecar.Stop(sidecarName2)).
// 3秒之后停止 app-2
Step("wait", flow.Sleep(3*time.Second)).
Step("stop app 2", app.Stop(appID2)).
// 等待30秒
Step("wait", flow.Sleep(30*time.Second)).
// 预期消息还是能全部收到（app-3/dapr-3还在工作）
Step("assert messages", assertMessages(consumerGroup2)).