아직 지원되지 않는다고 생각하지 마십시오. 이 JIRA 문제 “Add delete topic support”를 살펴보십시오 .

수동으로 삭제하려면 :

1. 클러스터 종료
2. kafka 로그 디렉토리 ( log.dirkafka 구성 파일 의 속성에 의해 지정됨 )와 사육사 데이터를 정리합니다.
3. 클러스터 다시 시작

주어진 주제에 대해 할 수있는 것은

1. 카프카 중지
2. 파티션에 특정한 kafka 로그를 정리합니다. kafka는 로그 파일을 “logDir / topic-partition”형식으로 저장하므로 “MyTopic”이라는 주제의 경우 파티션 ID 0에 대한 로그 가 속성으로 지정된 /tmp/kafka-logs/MyTopic-0위치에 저장됩니다 ./tmp/kafka-logslog.dir
3. kafka 다시 시작

이것은 NOT좋고 권장되는 접근 방식이지만 작동합니다. Kafka 브로커 구성 파일에서 log.retention.hours.per.topic속성은 다음을 정의하는 데 사용됩니다.The number of hours to keep a log file before deleting it for some specific topic

또한 소비자가 읽는 즉시 메시지가 삭제되는 방법이 있습니까?

로부터 카프카 문서 :

Kafka 클러스터는 사용 여부에 관계없이 게시 된 모든 메시지를 구성 가능한 기간 동안 유지합니다. 예를 들어 로그 보존이 2 일로 설정된 경우 메시지가 게시 된 후 2 일 동안 사용할 수 있으며 그 이후에는 공간을 확보하기 위해 폐기됩니다. Kafka의 성능은 데이터 크기와 관련하여 효과적으로 일정하므로 많은 데이터를 유지하는 것은 문제가되지 않습니다.

실제로 소비자별로 유지되는 유일한 메타 데이터는 “오프셋”이라고하는 로그에서 소비자의 위치입니다. 이 오프셋은 소비자에 의해 제어됩니다. 일반적으로 소비자는 메시지를 읽을 때 오프셋을 선형으로 진행하지만 실제로 위치는 소비자가 제어하며 원하는 순서대로 메시지를 소비 할 수 있습니다. 예를 들어 소비자는 이전 오프셋으로 재설정하여 재 처리 할 수 ​​있습니다.

시작을 찾기위한 카프카 0.8 읽어 오프셋 단순 소비자 예를 그들이 말하는을

Kafka는 도움이되는 두 개의 상수를 포함 kafka.api.OffsetRequest.EarliestTime()하고 있으며, 로그에서 데이터의 시작 부분을 찾고 거기에서 스트리밍을 시작하고 kafka.api.OffsetRequest.LatestTime()새 메시지 만 스트리밍합니다.

소비자 측에서 오프셋을 관리하기위한 예제 코드도 찾을 수 있습니다.

    public static long getLastOffset(SimpleConsumer consumer, String topic, int partition,
                                 long whichTime, String clientName) {
    TopicAndPartition topicAndPartition = new TopicAndPartition(topic, partition);
    Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo = new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
    requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(whichTime, 1));
    kafka.javaapi.OffsetRequest request = new kafka.javaapi.OffsetRequest(requestInfo, kafka.api.OffsetRequest.CurrentVersion(),clientName);
    OffsetResponse response = consumer.getOffsetsBefore(request);

    if (response.hasError()) {
        System.out.println("Error fetching data Offset Data the Broker. Reason: " + response.errorCode(topic, partition) );
        return 0;
    }
    long[] offsets = response.offsets(topic, partition);
    return offsets[0];
}