[jira] [Created] (KAFKA-6014) new consumer mirror maker halts after committing offsets to a deleted topic

["Onur Karaman (JIRA)" <ji...@apache.org>]

Onur Karaman created KAFKA-6014:
-----------------------------------

             Summary: new consumer mirror maker halts after committing offsets to a deleted topic
                 Key: KAFKA-6014
                 URL: https://issues.apache.org/jira/browse/KAFKA-6014
             Project: Kafka
          Issue Type: Bug
            Reporter: Onur Karaman


New consumer throws an unexpected KafkaException when trying to commit to a topic that has been deleted. MirrorMaker.commitOffsets doesn't attempt to catch the KafkaException and just kills the process. We didn't see this in the old consumer because old consumer just silently drops failed offset commits.

I ran a quick experiment locally to prove the behavior. The experiment:
1. start up a single broker
2. create a single-partition topic t
3. create a new consumer that consumes topic t
4. make the consumer commit every few seconds
5. delete topic t
6. expect: KafkaException that kills the process.

Here's my script:
{code}
package org.apache.kafka.clients.consumer;

import org.apache.kafka.common.TopicPartition;

import java.util.Collections;
import java.util.List;
import java.util.Properties;

public class OffsetCommitTopicDeletionTest {
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9090");
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "g");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        KafkaConsumer<byte[], byte[]> kafkaConsumer = new KafkaConsumer<>(props);
        TopicPartition partition = new TopicPartition("t", 0);
        List<TopicPartition> partitions = Collections.singletonList(partition);
        kafkaConsumer.assign(partitions);
        while (true) {
            kafkaConsumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(0, "")));
            Thread.sleep(1000);
        }
    }
}
{code}

Here are the other commands:
{code}
> rm -rf /tmp/zookeeper/ /tmp/kafka-logs* logs*
> ./gradlew clean jar
> ./bin/zookeeper-server-start.sh config/zookeeper.properties
> export LOG_DIR=logs0 && ./bin/kafka-server-start.sh config/server0.properties
> ./bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic t --partitions 1 --replication-factor 1
> ./bin/kafka-run-class.sh org.apache.kafka.clients.consumer.OffsetCommitTopicDeletionTest
> ./bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic t
{code}

Here is the output:
{code}
[2017-10-04 20:00:14,451] ERROR [Consumer clientId=consumer-1, groupId=g] Offset commit failed on partition t-0 at offset 0: This server does not host this topic-partition. (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)
Exception in thread "main" org.apache.kafka.common.KafkaException: Partition t-0 may not exist or user may not have Describe access to topic
  at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:789)
  at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:734)
  at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:808)
  at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:788)
  at org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:204)
  at org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:167)
  at org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:127)
  at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion(ConsumerNetworkClient.java:506)
  at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.firePendingCompletedRequests(ConsumerNetworkClient.java:353)
  at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:268)
  at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:214)
  at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:190)
  at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:600)
  at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1231)
  at org.apache.kafka.clients.consumer.OffsetCommitTopicDeletionTest.main(OffsetCommitTopicDeletionTest.java:22)
{code}

A couple ways we could fix this:
1. make OffsetCommitResponseHandler throw a more specific exception and make MirrorMaker.commitOffsets catch the exception. It currently just catches WakeupException and CommitFailedException.
2. make OffsetCommitResponseHandler log the error and move on. This is probably the simpler option. Just delete lines:
{code}
-                        future.raise(new KafkaException("Partition " + tp + " may not exist or user may not have Describe access to topic"));
-                        return;
{code}



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)