[ISSUE apache#4690] Fix Kafka Connector failed to start (apache#4691)

* Fix StringDeserializer & (session.timeout.ms)/3 > heartbeat.interval.ms
https://stackoverflow.com/questions/38406298/why-cant-i-increase-session-timeout-ms

* customize timeout config

* add new versions and remove redundant line-breaks

* Increase master version

* Fix an operator log typo

* Revert "Increase master version"

This reverts commit b112290.

* Revert 'add new versions for issue template'

Author: Pil0tXia

.github/PULL_REQUEST_TEMPLATE.md

@@ -30,14 +30,10 @@ Fixes #<XXXX>.
 *Explain the content here.*
 *Explain why you want to make the changes and what problem you're trying to solve.*
 
-
-
 ### Modifications
 
 *Describe the modifications you've done.*
 
-
-
 ### Documentation
 
 - Does this pull request introduce a new feature? (yes / no)

eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/source/config/SourceConnectorConfig.java

@@ -26,11 +26,11 @@ public class SourceConnectorConfig {
     private String topic = "TopicTest";
     private String bootstrapServers = "127.0.0.1:9092";
     private String groupID = "kafkaSource";
-    private String keyConverter = "org.apache.kafka.common.serialization.StringSerializer";
-    private String valueConverter = "org.apache.kafka.common.serialization.StringSerializer";
+    private String keyConverter = "org.apache.kafka.common.serialization.StringDeserializer";
+    private String valueConverter = "org.apache.kafka.common.serialization.StringDeserializer";
     private String autoCommitIntervalMS = "1000";
     private String enableAutoCommit = "false";
-    private String sessionTimeoutMS = "3000";
+    private String sessionTimeoutMS = "10000";
     private String maxPollRecords = "1000";
     private int pollTimeOut = 100;
 }

eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/source/connector/KafkaSourceConnector.java

@@ -76,7 +76,7 @@ private void doInit() {
         props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, sourceConfig.getConnectorConfig().getAutoCommitIntervalMS());
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sourceConfig.getConnectorConfig().getSessionTimeoutMS());
         this.pollTimeOut = sourceConfig.getConnectorConfig().getPollTimeOut();
-        this.kafkaConsumer = new KafkaConsumer<String, String>(props);
+        this.kafkaConsumer = new KafkaConsumer<>(props);
     }
 
     @Override

eventmesh-connectors/eventmesh-connector-kafka/src/main/resources/sink-config.yml

@@ -26,7 +26,7 @@ pubSubConfig:
     passWord: kafkaPassWord
 connectorConfig:
     connectorName: kafkaSink
-    bootstrapServers:
+    bootstrapServers: 127.0.0.1:9092
     topic: TopicTest
     keyConverter: org.apache.kafka.common.serialization.StringSerializer
     valueConverter: org.apache.kafka.common.serialization.StringSerializer

eventmesh-connectors/eventmesh-connector-kafka/src/main/resources/source-config.yml

@@ -26,7 +26,8 @@ pubSubConfig:
     passWord: kafkaPassWord
 connectorConfig:
     connectorName: kafkaSource
-    bootstrapServers: 127.0.0.1:9090
+    bootstrapServers: 127.0.0.1:9092
     topic: TopicTest
     groupID: kafkaSource
+    sessionTimeoutMS: 10000
     maxPollRecords: 1000

eventmesh-operator/controllers/eventmesh_connectors/connectors_controller.go

@@ -157,7 +157,7 @@ func (r ConnectorsReconciler) Reconcile(ctx context.Context, req reconcile.Reque
 		return reconcile.Result{}, err
 	}
 	podNames := getConnectorPodNames(podList.Items)
-	r.Logger.Info(fmt.Sprintf("Stutas.Nodes = %s", connector.Status.Nodes))
+	r.Logger.Info(fmt.Sprintf("Status.Nodes = %s", connector.Status.Nodes))
 	r.Logger.Info(fmt.Sprintf("podNames = %s", podNames))
 	// Ensure every pod is in running phase
 	for _, pod := range podList.Items {
@@ -168,7 +168,7 @@ func (r ConnectorsReconciler) Reconcile(ctx context.Context, req reconcile.Reque
 
 	if podNames != nil {
 		connector.Status.Nodes = podNames
-		r.Logger.Info(fmt.Sprintf("eventMeshRuntime.Stutas.Nodes = %s", connector.Status.Nodes))
+		r.Logger.Info(fmt.Sprintf("Connector.Status.Nodes = %s", connector.Status.Nodes))
 		// Update status.Size if needed
 		if connector.Spec.Size != connector.Status.Size {
 			r.Logger.Info("Connector.Status.Size = " + strconv.Itoa(connector.Status.Size))

eventmesh-operator/controllers/eventmesh_runtime/runtime_controller.go

@@ -197,7 +197,7 @@ func (r *RuntimeReconciler) Reconcile(ctx context.Context, req reconcile.Request
 	}
 
 	podNames := getRuntimePodNames(podList.Items)
-	r.Logger.Info(fmt.Sprintf("Stutas.Nodes = %s", eventMeshRuntime.Status.Nodes))
+	r.Logger.Info(fmt.Sprintf("Status.Nodes = %s", eventMeshRuntime.Status.Nodes))
 	r.Logger.Info(fmt.Sprintf("podNames = %s", podNames))
 	// Ensure every pod is in running phase
 	for _, pod := range podList.Items {
@@ -208,7 +208,7 @@ func (r *RuntimeReconciler) Reconcile(ctx context.Context, req reconcile.Request
 
 	if podNames != nil {
 		eventMeshRuntime.Status.Nodes = podNames
-		r.Logger.Info(fmt.Sprintf("eventMeshRuntime.Stutas.Nodes = %s", eventMeshRuntime.Status.Nodes))
+		r.Logger.Info(fmt.Sprintf("eventMeshRuntime.Status.Nodes = %s", eventMeshRuntime.Status.Nodes))
 		// Update status.Size if needed
 		if eventMeshRuntime.Spec.Size != eventMeshRuntime.Status.Size {
 			r.Logger.Info("eventMeshRuntime.Status.Size = " + strconv.Itoa(eventMeshRuntime.Status.Size))