Skip to content

fix: implement Exception Handling in Kafka Publisher #430

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 18 commits into
base: master
Choose a base branch
from
+80 −11
Open

fix: implement Exception Handling in Kafka Publisher #430

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
568c586
enhances the `PublisherServiceImpl `class by incorporating robust exc…
sancheet230 Mar 6, 2025
506b7d2
removing unwanted line
sancheet230 Mar 6, 2025
9659a97
Merge branch 'master' into issues
sancheet230 Mar 20, 2025
187e3c2
hardcode code changed
sancheet230 Mar 20, 2025
0ec7f46
reverting back requested code changes
sancheet230 Mar 20, 2025
b61a142
reverting back requested changes
sancheet230 Mar 20, 2025
c86af18
final requested chages
sancheet230 Mar 20, 2025
b81e5ed
Merge branch 'master' into issues
sancheet230 Jun 22, 2025
e3119b4
reverted required code for payload object of the message
sancheet230 Jun 22, 2025
2f13115
added and updated methods accordingly
sancheet230 Jun 22, 2025
5f53753
Update template for AsyncAPI Generator v2
sancheet230 Jun 22, 2025
073e115
Update dev dependency
sancheet230 Jun 22, 2025
003e39a
Update snapshots to match new generator output
sancheet230 Jun 22, 2025
2672620
Update kafka.test.js.snap
sancheet230 Jun 22, 2025
b5c8d5a
removed initially embedded cross and tick in error messages
sancheet230 Jun 26, 2025
e0e85a4
removed initially embedded cross and ticks from error messages
sancheet230 Jun 26, 2025
f973091
Updating back generator version
sancheet230 Jul 1, 2025
9486256
Updating back and editing max and min required generator version
sancheet230 Jul 1, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,7 @@
"required": false
}
},
"generator": ">=1.8.27 <2.0.0",
"generator": ">=1.8.27",
"filters": [
"@asyncapi/generator-filters"
],
Expand Down
5 changes: 3 additions & 2 deletions partials/KafkaPublisher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,9 @@ public interface PublisherService {
* {{line | safe}}{% endfor %}{% for line in channel.subscribe().description() | splitByLines %}
* {{line | safe}}{% endfor %}
*/{% endif %}
public void {{channel.subscribe().id() | camelCase}}(Integer key, {{varName | upperFirst}} {{varName}}{% if hasParameters %}{%for parameterName, parameter in channel.parameters() %}, {% if parameter.schema().type() === 'object'%}{{parameterName | camelCase | upperFirst}}{% else %}{{parameter.schema().type() | toJavaType(false)}}{% endif %} {{parameterName | camelCase}}{% endfor %}{% endif %});
public void {{methodName}}Async(Integer key, {{varName | upperFirst}} {{varName}}{% if hasParameters %}{%for parameterName, parameter in channel.parameters() %}, {% if parameter.schema().type() === 'object'%}{{parameterName | camelCase | upperFirst}}{% else %}{{parameter.schema().type() | toJavaType(false)}}{% endif %} {{parameterName | camelCase}}{% endfor %}{% endif %});
public void {{methodName}}Sync(Integer key, {{varName | upperFirst}} {{varName}}{% if hasParameters %}{%for parameterName, parameter in channel.parameters() %}, {% if parameter.schema().type() === 'object'%}{{parameterName | camelCase | upperFirst}}{% else %}{{parameter.schema().type() | toJavaType(false)}}{% endif %} {{parameterName | camelCase}}{% endfor %}{% endif %});
{%- endif %}
{%- endfor %}
}
{% endmacro %}
{% endmacro %}
43 changes: 38 additions & 5 deletions partials/KafkaPublisherImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,16 +13,22 @@
{%- endfor -%}
{% endif -%}
{% endfor %}

import javax.annotation.processing.Generated;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Generated(value="com.asyncapi.generator.template.spring", date="{{''|currentTime }}")
@Service
public class PublisherServiceImpl implements PublisherService {

@Autowired
private KafkaTemplate<Integer, Object> kafkaTemplate;

{% for channelName, channel in asyncapi.channels() %}
{%- if channel.hasSubscribe() %}
{%- set hasParameters = channel.hasParameters() %}
Expand All @@ -32,19 +38,45 @@ public class PublisherServiceImpl implements PublisherService {
{%- else %}
{%- set varName = channel.subscribe().message().payload().uid() | camelCase %}
{%- endif %}

{% if channel.description() or channel.subscribe().description() %}/**{% for line in channel.description() | splitByLines %}
* {{line | safe}}{% endfor %}{% for line in channel.subscribe().description() | splitByLines %}
* {{line | safe}}{% endfor %}
*/{% endif %}
public void {{methodName}}(Integer key, {{varName | upperFirst}} {{varName}}{% if hasParameters %}{%for parameterName, parameter in channel.parameters() %}, {% if parameter.schema().type() === 'object'%}{{parameterName | camelCase | upperFirst}}{% else %}{{parameter.schema().type() | toJavaType(false)}}{% endif %} {{parameterName | camelCase}}{% endfor %}{% endif %}) {
public void {{methodName}}Async(Integer key, {{varName | upperFirst}} {{varName}}{% if hasParameters %}{%for parameterName, parameter in channel.parameters() %}, {% if parameter.schema().type() === 'object'%}{{parameterName | camelCase | upperFirst}}{% else %}{{parameter.schema().type() | toJavaType(false)}}{% endif %} {{parameterName | camelCase}}{% endfor %}{% endif %}) {
Message<{{varName | upperFirst}}> message = MessageBuilder.withPayload({{varName}})
.setHeader(KafkaHeaders.TOPIC, get{{methodName | upperFirst-}}Topic({% if hasParameters %}{%for parameterName, parameter in channel.parameters() %}{{parameterName | camelCase}}{% if not loop.last %}, {% endif %}{% endfor %}{% endif %}))
.setHeader(KafkaHeaders.TOPIC, get{{methodName | upperFirst}}Topic({% if hasParameters %}{%for parameterName, parameter in channel.parameters() %}{{parameterName | camelCase}}{% if not loop.last %}, {% endif %}{% endfor %}{% endif %}))
.setHeader(KafkaHeaders.{%- if params.springBoot2 %}MESSAGE_KEY{% else %}KEY{% endif -%}, key)
.build();
kafkaTemplate.send(message);

CompletableFuture<?> future = kafkaTemplate.send(message);

future.whenComplete((result, ex) -> {
if (ex == null) {
System.out.println("Message successfully sent to topic: " + get{{methodName | upperFirst}}Topic({% if hasParameters %}{%for parameterName, parameter in channel.parameters() %}{{parameterName | camelCase}}{% if not loop.last %}, {% endif %}{% endfor %}{% endif %}) + " with key: " + key);
} else {
System.err.println("Failed to send message to topic: " + get{{methodName | upperFirst}}Topic({% if hasParameters %}{%for parameterName, parameter in channel.parameters() %}{{parameterName | camelCase}}{% if not loop.last %}, {% endif %}{% endfor %}{% endif %}) + " with key: " + key);
ex.printStackTrace();
}
});
}

private String get{{methodName | upperFirst-}}Topic({% if hasParameters %}{%for parameterName, parameter in channel.parameters() %}{% if parameter.schema().type() === 'object'%}{{parameterName | camelCase | upperFirst}}{% else %}{{parameter.schema().type() | toJavaType(false)}}{% endif %} {{parameterName | camelCase}}{% if not loop.last %}, {% endif %}{% endfor %}{% endif %}) {
public void {{methodName}}Sync(Integer key, {{varName | upperFirst}} {{varName}}{% if hasParameters %}{%for parameterName, parameter in channel.parameters() %}, {% if parameter.schema().type() === 'object'%}{{parameterName | camelCase | upperFirst}}{% else %}{{parameter.schema().type() | toJavaType(false)}}{% endif %} {{parameterName | camelCase}}{% endfor %}{% endif %}) {
Message<{{varName | upperFirst}}> message = MessageBuilder.withPayload({{varName}})
.setHeader(KafkaHeaders.TOPIC, get{{methodName | upperFirst}}Topic({% if hasParameters %}{%for parameterName, parameter in channel.parameters() %}{{parameterName | camelCase}}{% if not loop.last %}, {% endif %}{% endfor %}{% endif %}))
.setHeader(KafkaHeaders.{%- if params.springBoot2 %}MESSAGE_KEY{% else %}KEY{% endif -%}, key)
.build();

try {
kafkaTemplate.send(message).get(10, TimeUnit.SECONDS);
System.out.println("Message successfully sent synchronously to topic: " + get{{methodName | upperFirst}}Topic({% if hasParameters %}{%for parameterName, parameter in channel.parameters() %}{{parameterName | camelCase}}{% if not loop.last %}, {% endif %}{% endfor %}{% endif %}) + " with key: " + key);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
System.err.println("Failed to send message synchronously to topic: " + get{{methodName | upperFirst}}Topic({% if hasParameters %}{%for parameterName, parameter in channel.parameters() %}{{parameterName | camelCase}}{% if not loop.last %}, {% endif %}{% endfor %}{% endif %}) + " with key: " + key);
e.printStackTrace();
}
}

private String get{{methodName | upperFirst}}Topic({% if hasParameters %}{%for parameterName, parameter in channel.parameters() %}{% if parameter.schema().type() === 'object'%}{{parameterName | camelCase | upperFirst}}{% else %}{{parameter.schema().type() | toJavaType(false)}}{% endif %} {{parameterName | camelCase}}{% if not loop.last %}, {% endif %}{% endfor %}{% endif %}) {
Map<String, String> parameters = {% if hasParameters %}new HashMap<>(){% else %}null{% endif %};
{%- if hasParameters %}
{%- for parameterName, parameter in channel.parameters() %}
Expand All @@ -55,6 +87,7 @@ public class PublisherServiceImpl implements PublisherService {
}
{%- endif %}
{%- endfor %}

private String replaceParameters(String topic, Map<String, String> parameters) {
if (parameters != null) {
String compiledTopic = topic;
Expand All @@ -66,4 +99,4 @@ private String replaceParameters(String topic, Map<String, String> parameters) {
return topic;
}
}
{% endmacro %}
{% endmacro %}
41 changes: 38 additions & 3 deletions tests/__snapshots__/kafka.test.js.snap
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,7 +110,8 @@ public interface PublisherService {



public void updateLightMeasurement(Integer key, LightMeasuredPayload lightMeasuredPayload);
public void Async(Integer key, LightMeasuredPayload lightMeasuredPayload);
public void Sync(Integer key, LightMeasuredPayload lightMeasuredPayload);
}
"
`;
Expand Down Expand Up @@ -521,9 +522,14 @@ import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import com.asyncapi.model.LightMeasuredPayload;

import javax.annotation.processing.Generated;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Generated(value="com.asyncapi.generator.template.spring", date="AnyDate")
@Service
Expand All @@ -532,19 +538,47 @@ public class PublisherServiceImpl implements PublisherService {
@Autowired
private KafkaTemplate<Integer, Object> kafkaTemplate;




public void updateLightMeasurement(Integer key, LightMeasuredPayload lightMeasuredPayload) {
public void updateLightMeasurementAsync(Integer key, LightMeasuredPayload lightMeasuredPayload) {
Message<LightMeasuredPayload> message = MessageBuilder.withPayload(lightMeasuredPayload)
.setHeader(KafkaHeaders.TOPIC, getUpdateLightMeasurementTopic())
.setHeader(KafkaHeaders.KEY, key)
.build();

CompletableFuture<?> future = kafkaTemplate.send(message);

future.whenComplete((result, ex) -> {
if (ex == null) {
System.out.println("Message successfully sent to topic: " + getUpdateLightMeasurementTopic() + " with key: " + key);
} else {
System.err.println("Failed to send message to topic: " + getUpdateLightMeasurementTopic() + " with key: " + key);
ex.printStackTrace();
}
});
}

public void updateLightMeasurementSync(Integer key, LightMeasuredPayload lightMeasuredPayload) {
Message<LightMeasuredPayload> message = MessageBuilder.withPayload(lightMeasuredPayload)
.setHeader(KafkaHeaders.TOPIC, getUpdateLightMeasurementTopic())
.setHeader(KafkaHeaders.KEY, key)
.build();
kafkaTemplate.send(message);

try {
kafkaTemplate.send(message).get(10, TimeUnit.SECONDS);
System.out.println("Message successfully sent synchronously to topic: " + getUpdateLightMeasurementTopic() + " with key: " + key);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
System.err.println("Failed to send message synchronously to topic: " + getUpdateLightMeasurementTopic() + " with key: " + key);
e.printStackTrace();
}
}

private String getUpdateLightMeasurementTopic() {
Map<String, String> parameters = null;
return replaceParameters("event.lighting.measured", parameters);
}

private String replaceParameters(String topic, Map<String, String> parameters) {
if (parameters != null) {
String compiledTopic = topic;
Expand Down Expand Up @@ -725,3 +759,4 @@ public class MessageHandlerService {
}
"
`;