diff --git a/CHANGES.md b/CHANGES.md
index d6ecd6a7a..05e915048 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -17,6 +17,7 @@ Release Notes.
 * Fix apm-dubbo-2.7.x-plugin memory leak due to some Dubbo RpcExceptions.
 * Fix lettuce-5.x-plugin get null host in redis sentinel mode.
 * Fix ClassCastException by making CallbackAdapterInterceptor to implement EnhancedInstance interface in the spring-kafka plugin.
+* Fix NullPointerException with KafkaProducer.send(record).
 
 #### OAP-Backend
 * Allow user-defined `JAVA_OPTS` in the startup script.
diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackConstructorInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackConstructorInterceptor.java
index 7c13d9ec1..82cd4dce2 100644
--- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackConstructorInterceptor.java
+++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackConstructorInterceptor.java
@@ -30,8 +30,10 @@ public class CallbackConstructorInterceptor implements InstanceConstructorInterc
     @Override
     public void onConstruct(EnhancedInstance objInst, Object[] allArguments) {
         Callback callback = (Callback) allArguments[0];
-        CallbackCache cache = new CallbackCache();
-        cache.setCallback(callback);
-        objInst.setSkyWalkingDynamicField(cache);
+        if (null != callback) {
+            CallbackCache cache = new CallbackCache();
+            cache.setCallback(callback);
+            objInst.setSkyWalkingDynamicField(cache);
+        }
     }
 }
diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackInterceptor.java
index 888c330a6..03ac774c2 100644
--- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackInterceptor.java
+++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackInterceptor.java
@@ -74,7 +74,10 @@ public class CallbackInterceptor implements InstanceMethodsAroundInterceptor {
     @Override
     public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
         Class<?>[] argumentsTypes, Throwable t) {
-        ContextManager.activeSpan().log(t);
+        CallbackCache cache = (CallbackCache) objInst.getSkyWalkingDynamicField();
+        if (null != cache) {
+            ContextManager.activeSpan().log(t);
+        }
     }
 
     private ContextSnapshot getSnapshot(CallbackCache cache) {