3
\$\begingroup\$

I have two Kafka consumer ConsumerA and ConsumerB. I want to run these two kafka consumers independent of each other on the same machine. There is no relation between them at all. These two kafka consumer will work on different topics on the same machine.

  • Each consumer should have a different Properties object.
  • Each consumer should have a different thread pool configuration since they can be run in multithreaded way (consumer group) if needed independent of other consumer.

Here is my design:

Consumer class (abstract):

 public abstract class Consumer implements Runnable {
    private final Properties consumerProps;
    private final String consumerName;

    public Consumer(String consumerName, Properties consumerProps) {
        this.consumerName = consumerName;
        this.consumerProps = consumerProps;
    }

    protected abstract void shutdown();
    protected abstract void run(String consumerName, Properties consumerProps);

    @Override
    public final void run() {
        run(consumerName, consumerProps);
    }
}

ConsumerA class:

public class ConsumerA extends Consumer {
    private final AtomicBoolean closed = new AtomicBoolean(false);
    private KafkaConsumer<byte[], byte[]> consumer;

    public ConsumerA(String consumerName, Properties consumerProps) {
        super(consumerName, consumerProps);
    }

    @Override
    public void shutdown() {
        closed.set(true);
        consumer.wakeup();
    }

    @Override
    protected void run(String consumerName, Properties consumerProps) {
        consumer = new KafkaConsumer<>(consumerProps);
        consumer.subscribe(getTopicsBasisOnConsumerName());

        Map<String, Object> config = new HashMap<>();
        config.put(Config.URLS, TEST_URL);
        GenericRecordDomainDataDecoder decoder = new GenericRecordDomainDataDecoder(config);

        try {
            while (!closed.get()) {
                ConsumerRecords<byte[], byte[]> records = consumer.poll(Long.MAX_VALUE);
                for (ConsumerRecord<byte[], byte[]> record : records) {
                    GenericRecord payload = decoder.decode(record.value());
                    // extract data from payload
                    System.out.println("topic = %s, partition = %s, offset = %d, customer = %s, country = %s\n",
                                      record.topic(), record.partition(), record.offset(), record.key(), record.value());
                }
                consumer.commitAsync();
            }
        } catch (WakeupException ex) {
            // Ignore exception if closing
            System.out.println("error= ", ex);
            if (!closed.get()) throw e;             
        } catch (Exception ex) {
            System.out.println("error= ", ex);      
        } finally {
            try {
                consumer.commitSync();
            } finally {
                consumer.close();
            }
        }
    }
}

ConsumerA B class:

// similar to `ConsumerA` but with specific details of B

ConsumerHandler class:

public final class ConsumerHandler {
  private final ExecutorService executorServiceConsumer;
  private final Consumer consumer;
  private final List<Consumer> consumers = new ArrayList<>();

  public ConsumerHandler(Consumer consumer, int poolSize) {
    this.executorServiceConsumer = Executors.newFixedThreadPool(poolSize);
    this.consumer = consumer;
    for (int i = 0; i < poolSize; i++) {
      this.consumers.add(consumer);
      executorServiceConsumer.submit(consumer);
    }
 }
  public void shutdown() {
    Runtime.getRuntime().addShutdownHook(new Thread() {
      @Override
      public void run() {
        for (Consumer consumer : consumers) {
          consumer.shutdown();
        }
        executorServiceConsumer.shutdown();
        try {
          executorServiceConsumer.awaitTermination(1000, TimeUnit.MILLISECONDS);
        } catch (InterruptedException ex) {
          Thread.currentThread().interrupt();
        }
      }
    });
  }
}

Here is my main class in one of my project. If I start my server, calls will come first automatically, and from this place I start my all kafka consumers where I execute my ConsumerA and ConsumerB. And as soon as shutdown is called, I release all the resources by calling shutdown on all my Kafka consumers.

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.inject.Singleton;

@Singleton
@DependencyInjectionInitializer
public class Initializer {
  private ConsumerHandler consumerHandlerA;
  private ConsumerHandler consumerHandlerB;

  @PostConstruct
  public void init() {
    consumerHandlerA = new ConsumerHandler (new ConsumerA("consumerA", getConsumerPropsA()), 3);
    consumerHandlerB = new ConsumerHandler (new ConsumerB("consumerB", getConsumerPropsB()), 3);
  }

  @PreDestroy
  public void shutdown() {
    consumerHandlerA.shutdown();
    consumerHandlerB.shutdown();
  }
}

Is this the right design for this kind of problem where I want to run multiple kafka consumers on the same box? Let me know if there is any better and efficient way to solve this problem. In general I will be running three or four Kafka consumers max on the same box and each consumer can have their own consumer group if needed.

Here is the Javadoc for KafkaConsumer which I am using in both my consumer. And basis on this article I have created my consumer, it is just that I have used abstract class to extend it. Search for "Putting it all Together" in that link.

\$\endgroup\$
  • 1
    \$\begingroup\$ Have you considered running them in different JVMs? \$\endgroup\$ – h.j.k. Dec 18 '16 at 15:19
3
\$\begingroup\$

The majority of java coding standards recommend a 4-space indent (or a tabsize of 4 spaces). This code is indented with two.

</nitpicks>

Multithreading

The Docs you linked explicitly specify that Consumers are not Thread-Safe. While you're using the pattern from the docs there is a pretty big problem with how you use it:

  public ConsumerHandler(Consumer consumer, int poolSize) {
    this.executorServiceConsumer = Executors.newFixedThreadPool(poolSize);
    this.consumer = consumer;
    for (int i = 0; i < poolSize; i++) {
      this.consumers.add(consumer);
      executorServiceConsumer.submit(consumer);
    }
 }

Your code is reusing the same consumer instance for each thread in the pool. This is a bad idea, because calling the same method on the same instance of something is generally not considered threadsafe. Theoretically it should work just fine, but there is a problem that remains: You're calling shutdown on the same Consumer instance for each thread. That's theoretically not a major problem, because the method is written in a way that allows this.

But it's a waste. Since you're reusing the same instance anyways the following code in shutdown

    for (Consumer consumer : consumers) {
      consumer.shutdown();
    }

is equivalent to:

for (int i = 0; i < consumers.size(); i++) {
    consumer.shutdown(),
}

Additionally of note is the fact that ConsumerHandler#shutdown does not in fact shut down anything. It just registers a shutdown hook with the JVM. As such this method is:

  1. named incorrectly
  2. not idempotent (aka. can't be called multiple times without problems)
  3. inappropriately visible

I strongly suggest the following pattern instead, which allows you to properly shut down consumer threads, even when the JVM is not terminated:

public final class ConsumerHandler {
        private final ExecutorService executorServiceCustomer;
        private final Consumer consumer;
        private final Thread shutdownHook = new Thread() {
                @Override
                public void run() {
                        consumer.shutdown();
                        executorServiceCustomer.shutdown();
                }
        };

        // inside the constructor you should register the shutdownHook:
                Runtime.getRuntime().addShutdownHook(shutdownHook);

        // and inside the shutdown method,you should deregister it to allow finalization of dependent objects
        public void shutdown() {
                Runtime.getRuntime().removeShutdownHook(shutdownHook);
                shutdownHook.start();
                shutdownHook.join();
        }
}

Note that your shutdown hook is deregistered if it's not needed anymore. Additionally nobody can use your class to register any shutdown-hooks inadvertently and calling shutdown actually shuts the consumers down instead of having to wait for the JVM to terminate.

Interestingly your current implementation would only shut the consumers down when you call System.exit(). This is really undesired behaviour. That's the case because your ExecutorService does not use daemon threads, which means the JVM waits for the threads to terminate before exiting.

Timing

A final note about timing and waiting. Your shutdown hook can wait for up to a second before terminating. That's usually a bad sign. Instead of shutdown() and awaitTermination() you could've used shutdownNow().

Your consumers are extremely unlikely to free their assigned partition by themselves, since you gave them the astonishingly long time of almost 300 million years to report back to your cluster. Luckily the cluster doesn't feature quite that long timeouts between heartbeats, but you get the gist.

| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Posted a follow up question here by fixing thread safety issue as you mentioned. \$\endgroup\$ – david Dec 19 '16 at 6:37

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.