在Apache Beam中，可以使用KafkaIO作为Kafka消费者来读取消息。如果你的Kafka消费者一次又一次地重启，可能是由于以下几个原因引起的：

1. 程序中存在错误导致的异常：在处理消息的过程中，可能会发生异常导致程序崩溃并重启。确保你的代码中没有潜在的错误，比如空指针异常、数组越界等。

2. 网络问题：检查网络连接是否稳定，确保Kafka集群可以正常访问。

3. 配置问题：检查KafkaIO的配置是否正确。可能是配置中有错误或者缺少必要的配置项导致消费者重启。

下面是一个使用Apache Beam读取Kafka消息的代码示例，你可以参考这个示例来解决你的问题：

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;

public class KafkaConsumerExample {
  public static void main(String[] args) {
    // 创建PipelineOptions对象
    PipelineOptions options = PipelineOptionsFactory.create();

    // 创建Pipeline对象
    Pipeline pipeline = Pipeline.create(options);

    // Kafka消费者配置
    KafkaIO.Read kafkaReadOptions = KafkaIO.read()
        .withBootstrapServers("localhost:9092")
        .withTopic("my-topic")
        .withKeyDeserializer(StringDeserializer.class)
        .withValueDeserializer(LongDeserializer.class)
        .withoutMetadata();

    // 从Kafka读取消息
    pipeline.apply(KafkaIO.read()
        .withBootstrapServers("localhost:9092")
        .withTopic("my-topic")
        .withKeyDeserializer(StringDeserializer.class)
        .withValueDeserializer(LongDeserializer.class)
        .withoutMetadata())
        .apply(ParDo.of(new DoFn, Void>() {
          @ProcessElement
          public void processElement(ProcessContext c) {
            KafkaRecord record = c.element();
            // 处理消息
            System.out.println("Key: " + record.getKey() + ", Value: " + record.getValue());
          }
        }));

    // 运行Pipeline
    pipeline.run().waitUntilFinish();
  }
}

请注意，这只是一个示例代码，你需要根据你的实际情况进行适当的修改和调整。