This is an automated email from the ASF dual-hosted git repository.
mxsm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new fe7bf33a3 [ISSUE #4690] Fix Kafka Connector failed to start (#4691)
fe7bf33a3 is described below
commit fe7bf33a399beb572adb4f5cf7390ac0303651a3
Author: Pil0tXia <[email protected]>
AuthorDate: Thu Dec 28 16:30:59 2023 +0800
[ISSUE #4690] Fix Kafka Connector failed to start (#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 c60b5595c6ea2896b5f5c8009195e8b14a9bb3d6.
* Revert 'add new versions for issue template'
---
.github/PULL_REQUEST_TEMPLATE.md | 4 ----
.../connector/kafka/source/config/SourceConnectorConfig.java | 6 +++---
.../connector/kafka/source/connector/KafkaSourceConnector.java | 2 +-
.../eventmesh-connector-kafka/src/main/resources/sink-config.yml | 2 +-
.../eventmesh-connector-kafka/src/main/resources/source-config.yml | 3 ++-
.../controllers/eventmesh_connectors/connectors_controller.go | 4 ++--
.../controllers/eventmesh_runtime/runtime_controller.go | 4 ++--
7 files changed, 11 insertions(+), 14 deletions(-)
diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md
index 6595e7f6e..3ffc732e0 100644
--- a/.github/PULL_REQUEST_TEMPLATE.md
+++ b/.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)
diff --git
a/eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/source/config/SourceConnectorConfig.java
b/eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/source/config/SourceConnectorConfig.java
index c77bbda15..2d2f52f85 100644
---
a/eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/source/config/SourceConnectorConfig.java
+++
b/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;
}
diff --git
a/eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/source/connector/KafkaSourceConnector.java
b/eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/source/connector/KafkaSourceConnector.java
index 18fa42163..99fa73b58 100644
---
a/eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/source/connector/KafkaSourceConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/source/connector/KafkaSourceConnector.java
@@ -76,7 +76,7 @@ public class KafkaSourceConnector implements Source {
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
diff --git
a/eventmesh-connectors/eventmesh-connector-kafka/src/main/resources/sink-config.yml
b/eventmesh-connectors/eventmesh-connector-kafka/src/main/resources/sink-config.yml
index 241687825..b5c35b543 100644
---
a/eventmesh-connectors/eventmesh-connector-kafka/src/main/resources/sink-config.yml
+++
b/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
diff --git
a/eventmesh-connectors/eventmesh-connector-kafka/src/main/resources/source-config.yml
b/eventmesh-connectors/eventmesh-connector-kafka/src/main/resources/source-config.yml
index 4376df801..4300c749f 100644
---
a/eventmesh-connectors/eventmesh-connector-kafka/src/main/resources/source-config.yml
+++
b/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
diff --git
a/eventmesh-operator/controllers/eventmesh_connectors/connectors_controller.go
b/eventmesh-operator/controllers/eventmesh_connectors/connectors_controller.go
index d5aaa7e19..d554cbefe 100644
---
a/eventmesh-operator/controllers/eventmesh_connectors/connectors_controller.go
+++
b/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))
diff --git
a/eventmesh-operator/controllers/eventmesh_runtime/runtime_controller.go
b/eventmesh-operator/controllers/eventmesh_runtime/runtime_controller.go
index 89e433743..70aafefbf 100644
--- a/eventmesh-operator/controllers/eventmesh_runtime/runtime_controller.go
+++ b/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))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
