from django.conf import settings

from corehq.util.io import ClosingContextProxy
from kafka import KafkaConsumer
from kafka.client import KafkaClient, SimpleClient

GENERIC_KAFKA_CLIENT_ID = 'cchq-kafka-client'


def get_simple_kafka_client(client_id=GENERIC_KAFKA_CLIENT_ID):
    # this uses the old SimpleClient because we are using the old SimpleProducer interface
    return ClosingContextProxy(SimpleClient(
        hosts=settings.KAFKA_BROKERS,
        client_id=client_id,
        timeout=30,  # seconds
    ))


def get_kafka_client(client_id=GENERIC_KAFKA_CLIENT_ID):
    return ClosingContextProxy(KafkaClient(
        bootstrap_servers=settings.KAFKA_BROKERS,
        client_id=client_id,
        api_version=settings.KAFKA_API_VERSION
    ))


def get_kafka_consumer():
    return ClosingContextProxy(KafkaConsumer(
        client_id='pillowtop_utils',
        bootstrap_servers=settings.KAFKA_BROKERS,
        request_timeout_ms=1000
    ))

from django.conf import settings

from corehq.util.io import ClosingContextProxy
from kafka import KafkaConsumer
from kafka.client import KafkaClient, SimpleClient

GENERIC_KAFKA_CLIENT_ID = 'cchq-kafka-client'


def get_simple_kafka_client(client_id=GENERIC_KAFKA_CLIENT_ID):
    # this uses the old SimpleClient because we are using the old SimpleProducer interface
    return ClosingContextProxy(SimpleClient(
        hosts=settings.KAFKA_BROKERS,
        client_id=client_id,
        timeout=30,  # seconds
    ))


def get_kafka_client(client_id=GENERIC_KAFKA_CLIENT_ID):
    return ClosingContextProxy(KafkaClient(
        bootstrap_servers=settings.KAFKA_BROKERS,
        client_id=client_id,
        api_version=settings.KAFKA_API_VERSION
    ))


def get_kafka_consumer():
    return ClosingContextProxy(KafkaConsumer(
        client_id='pillowtop_utils',
        bootstrap_servers=settings.KAFKA_BROKERS,
    ))

Remove hard request timeout on kafka consumer
Remove hard request timeout on kafka consumer

  from django.conf import settings
  
  from corehq.util.io import ClosingContextProxy
  from kafka import KafkaConsumer
  from kafka.client import KafkaClient, SimpleClient
  
  GENERIC_KAFKA_CLIENT_ID = 'cchq-kafka-client'
  
  
  def get_simple_kafka_client(client_id=GENERIC_KAFKA_CLIENT_ID):
      # this uses the old SimpleClient because we are using the old SimpleProducer interface
      return ClosingContextProxy(SimpleClient(
          hosts=settings.KAFKA_BROKERS,
          client_id=client_id,
          timeout=30,  # seconds
      ))
  
  
  def get_kafka_client(client_id=GENERIC_KAFKA_CLIENT_ID):
      return ClosingContextProxy(KafkaClient(
          bootstrap_servers=settings.KAFKA_BROKERS,
          client_id=client_id,
          api_version=settings.KAFKA_API_VERSION
      ))
  
  
  def get_kafka_consumer():
      return ClosingContextProxy(KafkaConsumer(
          client_id='pillowtop_utils',
          bootstrap_servers=settings.KAFKA_BROKERS,
-         request_timeout_ms=1000
      ))
  
Remove hard request timeout on kafka consumer
## Code Before:
from django.conf import settings

from corehq.util.io import ClosingContextProxy
from kafka import KafkaConsumer
from kafka.client import KafkaClient, SimpleClient

GENERIC_KAFKA_CLIENT_ID = 'cchq-kafka-client'


def get_simple_kafka_client(client_id=GENERIC_KAFKA_CLIENT_ID):
    # this uses the old SimpleClient because we are using the old SimpleProducer interface
    return ClosingContextProxy(SimpleClient(
        hosts=settings.KAFKA_BROKERS,
        client_id=client_id,
        timeout=30,  # seconds
    ))


def get_kafka_client(client_id=GENERIC_KAFKA_CLIENT_ID):
    return ClosingContextProxy(KafkaClient(
        bootstrap_servers=settings.KAFKA_BROKERS,
        client_id=client_id,
        api_version=settings.KAFKA_API_VERSION
    ))


def get_kafka_consumer():
    return ClosingContextProxy(KafkaConsumer(
        client_id='pillowtop_utils',
        bootstrap_servers=settings.KAFKA_BROKERS,
        request_timeout_ms=1000
    ))

## Instruction:
Remove hard request timeout on kafka consumer
## Code After:
from django.conf import settings

from corehq.util.io import ClosingContextProxy
from kafka import KafkaConsumer
from kafka.client import KafkaClient, SimpleClient

GENERIC_KAFKA_CLIENT_ID = 'cchq-kafka-client'


def get_simple_kafka_client(client_id=GENERIC_KAFKA_CLIENT_ID):
    # this uses the old SimpleClient because we are using the old SimpleProducer interface
    return ClosingContextProxy(SimpleClient(
        hosts=settings.KAFKA_BROKERS,
        client_id=client_id,
        timeout=30,  # seconds
    ))


def get_kafka_client(client_id=GENERIC_KAFKA_CLIENT_ID):
    return ClosingContextProxy(KafkaClient(
        bootstrap_servers=settings.KAFKA_BROKERS,
        client_id=client_id,
        api_version=settings.KAFKA_API_VERSION
    ))


def get_kafka_consumer():
    return ClosingContextProxy(KafkaConsumer(
        client_id='pillowtop_utils',
        bootstrap_servers=settings.KAFKA_BROKERS,
    ))
