Apache Kafka是一个分布式流处理平台，它允许你发布和订阅数据流。以下是使用Java编写的Kafka生产者和消费者的简单示例，包括生产者的回调。

首先，确保你已经添加了Kafka客户端依赖到你的项目中。如果你使用的是Maven，可以在pom.xml文件中添加以下依赖：

<dependencies>
    <!-- Kafka client -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>3.1.0</version> <!-- 请使用最新的稳定版本 -->
    </dependency>
</dependencies>

以下是生产者的代码示例：

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        // 设置Kafka生产者的属性
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 创建Kafka生产者
        Producer<String, String> producer = new KafkaProducer<>(props);

        // 发送消息
        ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "key", "value");
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception e) {
                if (e != null) {
                    e.printStackTrace();
                } else {
                    System.out.println("The offset of the record we just sent is: " + metadata.offset());
                }
            }
        });

        // 关闭生产者
        producer.close();
    }
}

以下是消费者的代码示例：

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        // 设置Kafka消费者的属性
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        props.put("auto.offset.reset", "earliest");

        // 创建Kafka消费者
        Consumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅主题
        consumer.subscribe(Collections.singletonList("test-topic"));

        // 轮询消息
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                }
            }
        } finally {
            consumer.close();
        }
    }
}

请注意，你需要根据你的Kafka集群配置来修改bootstrap.servers、group.id和主题名称（在这个例子中是test-topic）。此外，确保你的Kafka集群正在运行，并且你有权限访问它。

在实际部署中，你可能需要处理更多的异常情况，并且可能需要根据你的业务逻辑来调整生产者和消费者的配置。这些代码仅作为一个简单的入门示例。