You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@kafka.apache.org by ankit tyagi <an...@gmail.com> on 2015/01/29 12:06:09 UTC
Potential socket leak in kafka sync producer
Hi,
Currently we are using sync producer client of 0.8.1 version in our
production box . we are getting the following exception while publishing
kafka message
*[2015-01-29
13:21:45.505][ThreadPoolTaskExecutor-603][WARN][ClientUtils$:89] Fetching
topic metadata with correlation id 10808 for topics [Set(*
*kafka_topic_coms_FD_test1)] from broker [id:0,host:localhost,port:9092]
failed*
*java.net.ConnectException: Connection refused*
* at sun.nio.ch.Net.connect0(Native Method)*
* at sun.nio.ch.Net.connect(Net.java:465)*
* at sun.nio.ch.Net.connect(Net.java:457)*
* at
sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:670)*
at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57)
at kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
at
kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:156)
at
kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68)
at kafka.producer.SyncProducer.send(SyncProducer.scala:112)
at
kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
at
kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
we are using dynamic thread pool to publish message to kafka. My
observation is when after keep alive time when threads in my executor gets
destroyed, somehow file descriptor is not getting cleared but when i did
explicitly ran the full gc, fd count got reduced by a signification amout.
Re: Potential socket leak in kafka sync producer
Posted by ankit tyagi <an...@gmail.com>.
attaching my producer whole code.
Creating kafkaProducer Bean in xml
* <bean id="kafkaESBProducer"
class="com.snapdeal.coms.kafka.KafkaProducer">*
* <constructor-arg name="topic" *
*
value="${KAFKA_ESB_TOPIC_NAME:kafka_topic_coms_esb_${COMS_PROFILE:dev}_${USER}}"
/>*
* <constructor-arg name="producerProperties">*
* <props>*
* <prop
key="metadata.broker.list">${KAFKA_PRODUCER_BROKER_LIST}</prop>*
* </props>*
* </constructor-arg>*
* </bean>*
public class KafkaProducer
{
private static final Logger LOG =
LoggerFactory.getLogger(KafkaProducer.class);
private static final String SYSTEM_USER_NAME_PROPERTY = "user.name";
private static final String CONFIG_PARAM_CLIENT_ID = "client.id";
private static final String CLIENT_ID_FORMAT_STR =
"kafka.coms.producer.%s.%s.%s";
private enum ConfigParam
{
SERIALIZER_CLASS("serializer.class",
CommonPropertyParam.KAFKA_PRODUCER_SERIALIZER_CLASS),
KEY_SERIALIZER_CLASS("key.serializer.class",
CommonPropertyParam.KAFKA_PRODUCER_PARTITION_KEY_SERIALIZER_CLASS),
//commenting this to use kafka default paritioner
// PARTITIONER_CLASS("partitioner.class",
// CommonPropertyParam.KAFKA_PRODUCER_PARTITIONER_CLASS),
REQUEST_REQUIRED_ACKS("request.required.acks",
CommonPropertyParam.KAFKA_PRODUCER_REQUEST_REQUIRED_ACKS);
private final String myName;
private final PropertyParam myParam;
ConfigParam(String name, PropertyParam param)
{
myName = name;
myParam = param;
}
public String getName()
{
return myName;
}
public PropertyParam getParam()
{
return myParam;
}
}
private final String myTopic;
private final Properties myProducerProperties;
private Producer<KafkaPartitionKey, KafkaEventWrapper> myProducer;
@Autowired
private COMSConfiguration myAppConfig;
public KafkaProducer(String topic, Properties producerProperties)
{
LOG.info("Creating Kafka Producer instance: {}", this);
myTopic = topic;
myProducerProperties = producerProperties;
}
* @PostConstruct*
* private void initializeProducer()*
* {*
* LOG.info("Initializing Kafka Producer for topic: {}", getTopic());*
* // Set producer unique client id*
* String currentUser =
System.getProperty(SYSTEM_USER_NAME_PROPERTY);*
* String currentJVMName =
ManagementFactory.getRuntimeMXBean().getName();*
* currentJVMName = currentJVMName.replace('@', '_');*
* String uniqueClientId = String.format(CLIENT_ID_FORMAT_STR,*
* getTopic(), currentUser, currentJVMName);*
* if (myProducerProperties.contains(CONFIG_PARAM_CLIENT_ID)) {*
* uniqueClientId += ":"*
* +
myProducerProperties.getProperty(CONFIG_PARAM_CLIENT_ID);*
* }*
* myProducerProperties.setProperty(CONFIG_PARAM_CLIENT_ID,*
* uniqueClientId);*
* // Set reasonable defaults for required params*
* for (ConfigParam cp : ConfigParam.values()) {*
* if (!myProducerProperties.containsKey(cp.getName())) {*
* String cpValue =
myAppConfig.getPropertyValue(cp.getParam());*
* myProducerProperties.setProperty(cp.getName(), cpValue);*
* }*
* }*
* myProducer =*
* new Producer<>(new ProducerConfig(myProducerProperties));*
* LOG.info("Initialized Kafka Producer for topic: {} and properties
{}", getTopic(),myProducerProperties);*
* }*
public String getTopic()
{
return myTopic;
}
public Producer<KafkaPartitionKey, KafkaEventWrapper> getProducer()
{
return myProducer;
}
public void send(KeyedMessage<KafkaPartitionKey, KafkaEventWrapper> msg)
{
myProducer.send(msg);
}
public void send(
List<KeyedMessage<KafkaPartitionKey, KafkaEventWrapper>> msgs)
{
myProducer.send(msgs);
}
* @PreDestroy*
* public void stop()*
* {*
* LOG.info("Stopping Kafka Producer for topic: {}", myTopic);*
* if (myProducer != null) {*
* myProducer.close();*
* }*
* }*
}
On Fri, Jan 30, 2015 at 1:08 PM, ankit tyagi <an...@gmail.com>
wrote:
> I have shared object histogram after and before gc on gist
> https://gist.github.com/ankit1987/f4a04a1350fdd609096d
>
> On Fri, Jan 30, 2015 at 12:43 PM, Jaikiran Pai <ja...@gmail.com>
> wrote:
>
>> What kind of a (managed) component is that which has the @PreDestroy?
>> Looking at the previous snippet you added, it looks like you are creating
>> the Producer in some method? If you are going to close the producer in a
>> @PreDestroy of the component, then you should be creating the producer in
>> the @PostConstruct of the same component, so that you have proper lifecycle
>> management of those resources.
>>
>>
>> -Jaikiran
>>
>> On Friday 30 January 2015 12:20 PM, ankit tyagi wrote:
>>
>>> Hi,
>>>
>>> I am closing my producer at the time of shutting down my application.
>>>
>>> @PreDestroy
>>> public void stop()
>>> {
>>> LOG.info("Stopping Kafka Producer for topic: {}", myTopic);
>>> if (myProducer != null) {
>>> myProducer.close();
>>> }
>>> }
>>>
>>>
>>>
>>> On Fri, Jan 30, 2015 at 11:22 AM, Manikumar Reddy <ku...@nmsworks.co.in>
>>> wrote:
>>>
>>> Hope you are closing the producers. can you share the attachment through
>>>> gist/patebin
>>>>
>>>> On Fri, Jan 30, 2015 at 11:11 AM, ankit tyagi <
>>>> ankittyagi.mnnit@gmail.com>
>>>> wrote:
>>>>
>>>> Hi Jaikiran,
>>>>>
>>>>> I am using ubuntu and was able to reproduce on redhat too. Please find
>>>>>
>>>> the
>>>>
>>>>> more information below.
>>>>>
>>>>>
>>>>> *DISTRIB_ID=Ubuntu*
>>>>> *DISTRIB_RELEASE=12.04*
>>>>> *DISTRIB_CODENAME=precise*
>>>>> *DISTRIB_DESCRIPTION="Ubuntu 12.04.5 LTS"*
>>>>>
>>>>> *java version "1.7.0_72"*
>>>>>
>>>>> This is happening on client side. Output of lsof was showing that
>>>>> maximum
>>>>> fd were FIFO and anon. But after GC FD count was reduced significantly.
>>>>>
>>>>> Below is my Client Code which i am using for publishing message.
>>>>>
>>>>>
>>>>> * private Producer<KafkaPartitionKey, KafkaEventWrapper> myProducer;*
>>>>>
>>>>> * myProducer = new Producer<>(new
>>>>> ProducerConfig(myProducerProperties));*
>>>>>
>>>>> * public void send(*
>>>>> * List<KeyedMessage<KafkaPartitionKey, KafkaEventWrapper>>
>>>>> msgs)*
>>>>> * {*
>>>>> * myProducer.send(msgs);*
>>>>> * }*
>>>>>
>>>>>
>>>>> we are using sync producer. I am attaching object histo before
>>>>>
>>>> GC(histo_1)
>>>>
>>>>> and after GC(histo_2) in my application.
>>>>>
>>>>> On Fri, Jan 30, 2015 at 9:34 AM, Jaikiran Pai <
>>>>> jai.forums2013@gmail.com>
>>>>> wrote:
>>>>>
>>>>> Which operating system are you on and what Java version? Depending on
>>>>>>
>>>>> the
>>>>
>>>>> OS, you could get tools (like lsof) to show which file descriptors are
>>>>>> being held on to. Is it the client JVM which ends up with these leaks?
>>>>>>
>>>>>> Also, would it be possible to post a snippet of your application code
>>>>>> which shows how you are using the Kafka APIs?
>>>>>>
>>>>>> -Jaikiran
>>>>>> On Thursday 29 January 2015 04:36 PM, ankit tyagi wrote:
>>>>>>
>>>>>> Hi,
>>>>>>>
>>>>>>> Currently we are using sync producer client of 0.8.1 version in our
>>>>>>> production box . we are getting the following exception while
>>>>>>>
>>>>>> publishing
>>>>
>>>>> kafka message
>>>>>>>
>>>>>>> *[2015-01-29
>>>>>>> 13:21:45.505][ThreadPoolTaskExecutor-603][WARN][ClientUtils$:89]
>>>>>>>
>>>>>> Fetching
>>>>>
>>>>>> topic metadata with correlation id 10808 for topics [Set(*
>>>>>>> *kafka_topic_coms_FD_test1)] from broker
>>>>>>>
>>>>>> [id:0,host:localhost,port:9092]
>>>>
>>>>> failed*
>>>>>>> *java.net.ConnectException: Connection refused*
>>>>>>> * at sun.nio.ch.Net.connect0(Native Method)*
>>>>>>> * at sun.nio.ch.Net.connect(Net.java:465)*
>>>>>>> * at sun.nio.ch.Net.connect(Net.java:457)*
>>>>>>> * at
>>>>>>> sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:670)*
>>>>>>> at
>>>>>>>
>>>>>> kafka.network.BlockingChannel.connect(BlockingChannel.scala:
>>>>
>>>>> 57)
>>>>>>> at
>>>>>>>
>>>>>> kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
>>>>
>>>>> at
>>>>>>>
>>>>>>> kafka.producer.SyncProducer.getOrMakeConnection(
>>>> SyncProducer.scala:156)
>>>>
>>>>> at
>>>>>>> kafka.producer.SyncProducer.kafka$producer$SyncProducer$$
>>>>>>> doSend(SyncProducer.scala:68)
>>>>>>> at kafka.producer.SyncProducer.
>>>>>>> send(SyncProducer.scala:112)
>>>>>>> at
>>>>>>> kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
>>>>>>> at
>>>>>>> kafka.producer.BrokerPartitionInfo.updateInfo(
>>>>>>> BrokerPartitionInfo.scala:82)
>>>>>>>
>>>>>>>
>>>>>>> we are using dynamic thread pool to publish message to kafka. My
>>>>>>> observation is when after keep alive time when threads in my executor
>>>>>>>
>>>>>> gets
>>>>>
>>>>>> destroyed, somehow file descriptor is not getting cleared but when i
>>>>>>>
>>>>>> did
>>>>
>>>>> explicitly ran the full gc, fd count got reduced by a signification
>>>>>>>
>>>>>> amout.
>>>>>
>>>>>>
>>>>>>>
>>
>
Re: Potential socket leak in kafka sync producer
Posted by Jaikiran Pai <ja...@gmail.com>.
Hi Ankit,
Would you be able to share the trimmed down application code which
reproduces this issue (maybe as a repo on github)? That way, some of us
will get more context about the issue and probably be able to try it out
ourselves and see what's wrong.
On a related note, have you tried this against a later version (like the
0.8.2.0 CR3) and see if it's still reproducible there?
-Jaikiran
On Friday 30 January 2015 02:59 PM, ankit tyagi wrote:
> Jaikiran,
>
> I have already investigated that this is kafka related. I made a small
> application which is used only for publishing messages to kafka. If I use
> dynamic thread pool means where number of maxPoolSize is very large
> comparative to corePoolSize and I publish each batch of messages only after
> all threads gets destroyed after keepAliveSeconds then FD leak problem
> occurs . I suspect when threads gets destroyed some how file handlers is
> not getting cleared . So when I trigger explict GC, descriptor count get
> reduced by signifact amont because of clean up of those destroyed thread.
>
>
> we got this problem in our production box where soft and hard limit of file
> descriptor was 50000 but for reproducing this issue on my local machine i
> have reduced hard limit to 6000 and used 1000 threads to send message to
> kafka (topic had 100 partition with 1 replication factor)
>
>
>
>
>
> On Fri, Jan 30, 2015 at 2:14 PM, Jaikiran Pai <ja...@gmail.com>
> wrote:
>
>> Looking at that heap dump, this probably is a database connection/resource
>> leak (298 connections?) than anything to do with Kafka. Have you
>> investigated if there's any DB resource leak in the application and ruled
>> out that part?
>>
>> -Jaikiran
>>
>>
>> On Friday 30 January 2015 01:08 PM, ankit tyagi wrote:
>>
>>> I have shared object histogram after and before gc on gist
>>> https://gist.github.com/ankit1987/f4a04a1350fdd609096d
>>>
>>> On Fri, Jan 30, 2015 at 12:43 PM, Jaikiran Pai <ja...@gmail.com>
>>> wrote:
>>>
>>> What kind of a (managed) component is that which has the @PreDestroy?
>>>> Looking at the previous snippet you added, it looks like you are creating
>>>> the Producer in some method? If you are going to close the producer in a
>>>> @PreDestroy of the component, then you should be creating the producer in
>>>> the @PostConstruct of the same component, so that you have proper
>>>> lifecycle
>>>> management of those resources.
>>>>
>>>>
>>>> -Jaikiran
>>>>
>>>> On Friday 30 January 2015 12:20 PM, ankit tyagi wrote:
>>>>
>>>> Hi,
>>>>> I am closing my producer at the time of shutting down my application.
>>>>>
>>>>> @PreDestroy
>>>>> public void stop()
>>>>> {
>>>>> LOG.info("Stopping Kafka Producer for topic: {}", myTopic);
>>>>> if (myProducer != null) {
>>>>> myProducer.close();
>>>>> }
>>>>> }
>>>>>
>>>>>
>>>>>
>>>>> On Fri, Jan 30, 2015 at 11:22 AM, Manikumar Reddy <kumar@nmsworks.co.in
>>>>> wrote:
>>>>>
>>>>> Hope you are closing the producers. can you share the attachment
>>>>> through
>>>>>
>>>>>> gist/patebin
>>>>>>
>>>>>> On Fri, Jan 30, 2015 at 11:11 AM, ankit tyagi <
>>>>>> ankittyagi.mnnit@gmail.com>
>>>>>> wrote:
>>>>>>
>>>>>> Hi Jaikiran,
>>>>>>
>>>>>>> I am using ubuntu and was able to reproduce on redhat too. Please find
>>>>>>>
>>>>>>> the
>>>>>> more information below.
>>>>>>>
>>>>>>> *DISTRIB_ID=Ubuntu*
>>>>>>> *DISTRIB_RELEASE=12.04*
>>>>>>> *DISTRIB_CODENAME=precise*
>>>>>>> *DISTRIB_DESCRIPTION="Ubuntu 12.04.5 LTS"*
>>>>>>>
>>>>>>> *java version "1.7.0_72"*
>>>>>>>
>>>>>>> This is happening on client side. Output of lsof was showing that
>>>>>>> maximum
>>>>>>> fd were FIFO and anon. But after GC FD count was reduced
>>>>>>> significantly.
>>>>>>>
>>>>>>> Below is my Client Code which i am using for publishing message.
>>>>>>>
>>>>>>>
>>>>>>> * private Producer<KafkaPartitionKey, KafkaEventWrapper> myProducer;*
>>>>>>>
>>>>>>> * myProducer = new Producer<>(new
>>>>>>> ProducerConfig(myProducerProperties));*
>>>>>>>
>>>>>>> * public void send(*
>>>>>>> * List<KeyedMessage<KafkaPartitionKey, KafkaEventWrapper>>
>>>>>>> msgs)*
>>>>>>> * {*
>>>>>>> * myProducer.send(msgs);*
>>>>>>> * }*
>>>>>>>
>>>>>>>
>>>>>>> we are using sync producer. I am attaching object histo before
>>>>>>>
>>>>>>> GC(histo_1)
>>>>>> and after GC(histo_2) in my application.
>>>>>>> On Fri, Jan 30, 2015 at 9:34 AM, Jaikiran Pai <
>>>>>>> jai.forums2013@gmail.com
>>>>>>> wrote:
>>>>>>>
>>>>>>> Which operating system are you on and what Java version? Depending
>>>>>>> on
>>>>>>> the
>>>>>>> OS, you could get tools (like lsof) to show which file descriptors are
>>>>>>>
>>>>>>>> being held on to. Is it the client JVM which ends up with these
>>>>>>>> leaks?
>>>>>>>>
>>>>>>>> Also, would it be possible to post a snippet of your application code
>>>>>>>> which shows how you are using the Kafka APIs?
>>>>>>>>
>>>>>>>> -Jaikiran
>>>>>>>> On Thursday 29 January 2015 04:36 PM, ankit tyagi wrote:
>>>>>>>>
>>>>>>>> Hi,
>>>>>>>>
>>>>>>>>> Currently we are using sync producer client of 0.8.1 version in our
>>>>>>>>> production box . we are getting the following exception while
>>>>>>>>>
>>>>>>>>> publishing
>>>>>>> kafka message
>>>>>>>
>>>>>>>> *[2015-01-29
>>>>>>>>> 13:21:45.505][ThreadPoolTaskExecutor-603][WARN][ClientUtils$:89]
>>>>>>>>>
>>>>>>>>> Fetching
>>>>>>>> topic metadata with correlation id 10808 for topics [Set(*
>>>>>>>>
>>>>>>>>> *kafka_topic_coms_FD_test1)] from broker
>>>>>>>>>
>>>>>>>>> [id:0,host:localhost,port:9092]
>>>>>>> failed*
>>>>>>>
>>>>>>>> *java.net.ConnectException: Connection refused*
>>>>>>>>> * at sun.nio.ch.Net.connect0(Native Method)*
>>>>>>>>> * at sun.nio.ch.Net.connect(Net.java:465)*
>>>>>>>>> * at sun.nio.ch.Net.connect(Net.java:457)*
>>>>>>>>> * at
>>>>>>>>> sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:670)*
>>>>>>>>> at
>>>>>>>>>
>>>>>>>>> kafka.network.BlockingChannel.connect(BlockingChannel.scala:
>>>>>>> 57)
>>>>>>>
>>>>>>>> at
>>>>>>>>> kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
>>>>>>> at
>>>>>>>
>>>>>>>> kafka.producer.SyncProducer.getOrMakeConnection(
>>>>>>>> SyncProducer.scala:156)
>>>>>> at
>>>>>>>> kafka.producer.SyncProducer.kafka$producer$SyncProducer$$
>>>>>>>>> doSend(SyncProducer.scala:68)
>>>>>>>>> at kafka.producer.SyncProducer.
>>>>>>>>> send(SyncProducer.scala:112)
>>>>>>>>> at
>>>>>>>>> kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
>>>>>>>>> at
>>>>>>>>> kafka.producer.BrokerPartitionInfo.updateInfo(
>>>>>>>>> BrokerPartitionInfo.scala:82)
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> we are using dynamic thread pool to publish message to kafka. My
>>>>>>>>> observation is when after keep alive time when threads in my
>>>>>>>>> executor
>>>>>>>>>
>>>>>>>>> gets
>>>>>>>> destroyed, somehow file descriptor is not getting cleared but when i
>>>>>>>> did
>>>>>>>>
>>>>>>> explicitly ran the full gc, fd count got reduced by a signification
>>>>>>>
>>>>>>>> amout.
>>>>>>>>
Re: Potential socket leak in kafka sync producer
Posted by ankit tyagi <an...@gmail.com>.
Jaikiran,
I have already investigated that this is kafka related. I made a small
application which is used only for publishing messages to kafka. If I use
dynamic thread pool means where number of maxPoolSize is very large
comparative to corePoolSize and I publish each batch of messages only after
all threads gets destroyed after keepAliveSeconds then FD leak problem
occurs . I suspect when threads gets destroyed some how file handlers is
not getting cleared . So when I trigger explict GC, descriptor count get
reduced by signifact amont because of clean up of those destroyed thread.
we got this problem in our production box where soft and hard limit of file
descriptor was 50000 but for reproducing this issue on my local machine i
have reduced hard limit to 6000 and used 1000 threads to send message to
kafka (topic had 100 partition with 1 replication factor)
On Fri, Jan 30, 2015 at 2:14 PM, Jaikiran Pai <ja...@gmail.com>
wrote:
> Looking at that heap dump, this probably is a database connection/resource
> leak (298 connections?) than anything to do with Kafka. Have you
> investigated if there's any DB resource leak in the application and ruled
> out that part?
>
> -Jaikiran
>
>
> On Friday 30 January 2015 01:08 PM, ankit tyagi wrote:
>
>> I have shared object histogram after and before gc on gist
>> https://gist.github.com/ankit1987/f4a04a1350fdd609096d
>>
>> On Fri, Jan 30, 2015 at 12:43 PM, Jaikiran Pai <ja...@gmail.com>
>> wrote:
>>
>> What kind of a (managed) component is that which has the @PreDestroy?
>>> Looking at the previous snippet you added, it looks like you are creating
>>> the Producer in some method? If you are going to close the producer in a
>>> @PreDestroy of the component, then you should be creating the producer in
>>> the @PostConstruct of the same component, so that you have proper
>>> lifecycle
>>> management of those resources.
>>>
>>>
>>> -Jaikiran
>>>
>>> On Friday 30 January 2015 12:20 PM, ankit tyagi wrote:
>>>
>>> Hi,
>>>>
>>>> I am closing my producer at the time of shutting down my application.
>>>>
>>>> @PreDestroy
>>>> public void stop()
>>>> {
>>>> LOG.info("Stopping Kafka Producer for topic: {}", myTopic);
>>>> if (myProducer != null) {
>>>> myProducer.close();
>>>> }
>>>> }
>>>>
>>>>
>>>>
>>>> On Fri, Jan 30, 2015 at 11:22 AM, Manikumar Reddy <kumar@nmsworks.co.in
>>>> >
>>>> wrote:
>>>>
>>>> Hope you are closing the producers. can you share the attachment
>>>> through
>>>>
>>>>> gist/patebin
>>>>>
>>>>> On Fri, Jan 30, 2015 at 11:11 AM, ankit tyagi <
>>>>> ankittyagi.mnnit@gmail.com>
>>>>> wrote:
>>>>>
>>>>> Hi Jaikiran,
>>>>>
>>>>>> I am using ubuntu and was able to reproduce on redhat too. Please find
>>>>>>
>>>>>> the
>>>>>
>>>>> more information below.
>>>>>>
>>>>>>
>>>>>> *DISTRIB_ID=Ubuntu*
>>>>>> *DISTRIB_RELEASE=12.04*
>>>>>> *DISTRIB_CODENAME=precise*
>>>>>> *DISTRIB_DESCRIPTION="Ubuntu 12.04.5 LTS"*
>>>>>>
>>>>>> *java version "1.7.0_72"*
>>>>>>
>>>>>> This is happening on client side. Output of lsof was showing that
>>>>>> maximum
>>>>>> fd were FIFO and anon. But after GC FD count was reduced
>>>>>> significantly.
>>>>>>
>>>>>> Below is my Client Code which i am using for publishing message.
>>>>>>
>>>>>>
>>>>>> * private Producer<KafkaPartitionKey, KafkaEventWrapper> myProducer;*
>>>>>>
>>>>>> * myProducer = new Producer<>(new
>>>>>> ProducerConfig(myProducerProperties));*
>>>>>>
>>>>>> * public void send(*
>>>>>> * List<KeyedMessage<KafkaPartitionKey, KafkaEventWrapper>>
>>>>>> msgs)*
>>>>>> * {*
>>>>>> * myProducer.send(msgs);*
>>>>>> * }*
>>>>>>
>>>>>>
>>>>>> we are using sync producer. I am attaching object histo before
>>>>>>
>>>>>> GC(histo_1)
>>>>>
>>>>> and after GC(histo_2) in my application.
>>>>>>
>>>>>> On Fri, Jan 30, 2015 at 9:34 AM, Jaikiran Pai <
>>>>>> jai.forums2013@gmail.com
>>>>>> wrote:
>>>>>>
>>>>>> Which operating system are you on and what Java version? Depending
>>>>>> on
>>>>>> the
>>>>>> OS, you could get tools (like lsof) to show which file descriptors are
>>>>>>
>>>>>>> being held on to. Is it the client JVM which ends up with these
>>>>>>> leaks?
>>>>>>>
>>>>>>> Also, would it be possible to post a snippet of your application code
>>>>>>> which shows how you are using the Kafka APIs?
>>>>>>>
>>>>>>> -Jaikiran
>>>>>>> On Thursday 29 January 2015 04:36 PM, ankit tyagi wrote:
>>>>>>>
>>>>>>> Hi,
>>>>>>>
>>>>>>>> Currently we are using sync producer client of 0.8.1 version in our
>>>>>>>> production box . we are getting the following exception while
>>>>>>>>
>>>>>>>> publishing
>>>>>>>
>>>>>> kafka message
>>>>>>
>>>>>>> *[2015-01-29
>>>>>>>> 13:21:45.505][ThreadPoolTaskExecutor-603][WARN][ClientUtils$:89]
>>>>>>>>
>>>>>>>> Fetching
>>>>>>> topic metadata with correlation id 10808 for topics [Set(*
>>>>>>>
>>>>>>>> *kafka_topic_coms_FD_test1)] from broker
>>>>>>>>
>>>>>>>> [id:0,host:localhost,port:9092]
>>>>>>>
>>>>>> failed*
>>>>>>
>>>>>>> *java.net.ConnectException: Connection refused*
>>>>>>>> * at sun.nio.ch.Net.connect0(Native Method)*
>>>>>>>> * at sun.nio.ch.Net.connect(Net.java:465)*
>>>>>>>> * at sun.nio.ch.Net.connect(Net.java:457)*
>>>>>>>> * at
>>>>>>>> sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:670)*
>>>>>>>> at
>>>>>>>>
>>>>>>>> kafka.network.BlockingChannel.connect(BlockingChannel.scala:
>>>>>>>
>>>>>> 57)
>>>>>>
>>>>>>> at
>>>>>>>>
>>>>>>>> kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
>>>>>>>
>>>>>> at
>>>>>>
>>>>>>> kafka.producer.SyncProducer.getOrMakeConnection(
>>>>>>>>
>>>>>>> SyncProducer.scala:156)
>>>>>
>>>>> at
>>>>>>
>>>>>>> kafka.producer.SyncProducer.kafka$producer$SyncProducer$$
>>>>>>>> doSend(SyncProducer.scala:68)
>>>>>>>> at kafka.producer.SyncProducer.
>>>>>>>> send(SyncProducer.scala:112)
>>>>>>>> at
>>>>>>>> kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
>>>>>>>> at
>>>>>>>> kafka.producer.BrokerPartitionInfo.updateInfo(
>>>>>>>> BrokerPartitionInfo.scala:82)
>>>>>>>>
>>>>>>>>
>>>>>>>> we are using dynamic thread pool to publish message to kafka. My
>>>>>>>> observation is when after keep alive time when threads in my
>>>>>>>> executor
>>>>>>>>
>>>>>>>> gets
>>>>>>> destroyed, somehow file descriptor is not getting cleared but when i
>>>>>>> did
>>>>>>>
>>>>>> explicitly ran the full gc, fd count got reduced by a signification
>>>>>>
>>>>>>> amout.
>>>>>>>
>>>>>>
>
Re: Potential socket leak in kafka sync producer
Posted by Jaikiran Pai <ja...@gmail.com>.
Looking at that heap dump, this probably is a database
connection/resource leak (298 connections?) than anything to do with
Kafka. Have you investigated if there's any DB resource leak in the
application and ruled out that part?
-Jaikiran
On Friday 30 January 2015 01:08 PM, ankit tyagi wrote:
> I have shared object histogram after and before gc on gist
> https://gist.github.com/ankit1987/f4a04a1350fdd609096d
>
> On Fri, Jan 30, 2015 at 12:43 PM, Jaikiran Pai <ja...@gmail.com>
> wrote:
>
>> What kind of a (managed) component is that which has the @PreDestroy?
>> Looking at the previous snippet you added, it looks like you are creating
>> the Producer in some method? If you are going to close the producer in a
>> @PreDestroy of the component, then you should be creating the producer in
>> the @PostConstruct of the same component, so that you have proper lifecycle
>> management of those resources.
>>
>>
>> -Jaikiran
>>
>> On Friday 30 January 2015 12:20 PM, ankit tyagi wrote:
>>
>>> Hi,
>>>
>>> I am closing my producer at the time of shutting down my application.
>>>
>>> @PreDestroy
>>> public void stop()
>>> {
>>> LOG.info("Stopping Kafka Producer for topic: {}", myTopic);
>>> if (myProducer != null) {
>>> myProducer.close();
>>> }
>>> }
>>>
>>>
>>>
>>> On Fri, Jan 30, 2015 at 11:22 AM, Manikumar Reddy <ku...@nmsworks.co.in>
>>> wrote:
>>>
>>> Hope you are closing the producers. can you share the attachment through
>>>> gist/patebin
>>>>
>>>> On Fri, Jan 30, 2015 at 11:11 AM, ankit tyagi <
>>>> ankittyagi.mnnit@gmail.com>
>>>> wrote:
>>>>
>>>> Hi Jaikiran,
>>>>> I am using ubuntu and was able to reproduce on redhat too. Please find
>>>>>
>>>> the
>>>>
>>>>> more information below.
>>>>>
>>>>>
>>>>> *DISTRIB_ID=Ubuntu*
>>>>> *DISTRIB_RELEASE=12.04*
>>>>> *DISTRIB_CODENAME=precise*
>>>>> *DISTRIB_DESCRIPTION="Ubuntu 12.04.5 LTS"*
>>>>>
>>>>> *java version "1.7.0_72"*
>>>>>
>>>>> This is happening on client side. Output of lsof was showing that
>>>>> maximum
>>>>> fd were FIFO and anon. But after GC FD count was reduced significantly.
>>>>>
>>>>> Below is my Client Code which i am using for publishing message.
>>>>>
>>>>>
>>>>> * private Producer<KafkaPartitionKey, KafkaEventWrapper> myProducer;*
>>>>>
>>>>> * myProducer = new Producer<>(new
>>>>> ProducerConfig(myProducerProperties));*
>>>>>
>>>>> * public void send(*
>>>>> * List<KeyedMessage<KafkaPartitionKey, KafkaEventWrapper>>
>>>>> msgs)*
>>>>> * {*
>>>>> * myProducer.send(msgs);*
>>>>> * }*
>>>>>
>>>>>
>>>>> we are using sync producer. I am attaching object histo before
>>>>>
>>>> GC(histo_1)
>>>>
>>>>> and after GC(histo_2) in my application.
>>>>>
>>>>> On Fri, Jan 30, 2015 at 9:34 AM, Jaikiran Pai <jai.forums2013@gmail.com
>>>>> wrote:
>>>>>
>>>>> Which operating system are you on and what Java version? Depending on
>>>>> the
>>>>> OS, you could get tools (like lsof) to show which file descriptors are
>>>>>> being held on to. Is it the client JVM which ends up with these leaks?
>>>>>>
>>>>>> Also, would it be possible to post a snippet of your application code
>>>>>> which shows how you are using the Kafka APIs?
>>>>>>
>>>>>> -Jaikiran
>>>>>> On Thursday 29 January 2015 04:36 PM, ankit tyagi wrote:
>>>>>>
>>>>>> Hi,
>>>>>>> Currently we are using sync producer client of 0.8.1 version in our
>>>>>>> production box . we are getting the following exception while
>>>>>>>
>>>>>> publishing
>>>>> kafka message
>>>>>>> *[2015-01-29
>>>>>>> 13:21:45.505][ThreadPoolTaskExecutor-603][WARN][ClientUtils$:89]
>>>>>>>
>>>>>> Fetching
>>>>>> topic metadata with correlation id 10808 for topics [Set(*
>>>>>>> *kafka_topic_coms_FD_test1)] from broker
>>>>>>>
>>>>>> [id:0,host:localhost,port:9092]
>>>>> failed*
>>>>>>> *java.net.ConnectException: Connection refused*
>>>>>>> * at sun.nio.ch.Net.connect0(Native Method)*
>>>>>>> * at sun.nio.ch.Net.connect(Net.java:465)*
>>>>>>> * at sun.nio.ch.Net.connect(Net.java:457)*
>>>>>>> * at
>>>>>>> sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:670)*
>>>>>>> at
>>>>>>>
>>>>>> kafka.network.BlockingChannel.connect(BlockingChannel.scala:
>>>>> 57)
>>>>>>> at
>>>>>>>
>>>>>> kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
>>>>> at
>>>>>>> kafka.producer.SyncProducer.getOrMakeConnection(
>>>> SyncProducer.scala:156)
>>>>
>>>>> at
>>>>>>> kafka.producer.SyncProducer.kafka$producer$SyncProducer$$
>>>>>>> doSend(SyncProducer.scala:68)
>>>>>>> at kafka.producer.SyncProducer.send(SyncProducer.scala:112)
>>>>>>> at
>>>>>>> kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
>>>>>>> at
>>>>>>> kafka.producer.BrokerPartitionInfo.updateInfo(
>>>>>>> BrokerPartitionInfo.scala:82)
>>>>>>>
>>>>>>>
>>>>>>> we are using dynamic thread pool to publish message to kafka. My
>>>>>>> observation is when after keep alive time when threads in my executor
>>>>>>>
>>>>>> gets
>>>>>> destroyed, somehow file descriptor is not getting cleared but when i
>>>>>> did
>>>>> explicitly ran the full gc, fd count got reduced by a signification
>>>>>> amout.
Re: Potential socket leak in kafka sync producer
Posted by ankit tyagi <an...@gmail.com>.
I have shared object histogram after and before gc on gist
https://gist.github.com/ankit1987/f4a04a1350fdd609096d
On Fri, Jan 30, 2015 at 12:43 PM, Jaikiran Pai <ja...@gmail.com>
wrote:
> What kind of a (managed) component is that which has the @PreDestroy?
> Looking at the previous snippet you added, it looks like you are creating
> the Producer in some method? If you are going to close the producer in a
> @PreDestroy of the component, then you should be creating the producer in
> the @PostConstruct of the same component, so that you have proper lifecycle
> management of those resources.
>
>
> -Jaikiran
>
> On Friday 30 January 2015 12:20 PM, ankit tyagi wrote:
>
>> Hi,
>>
>> I am closing my producer at the time of shutting down my application.
>>
>> @PreDestroy
>> public void stop()
>> {
>> LOG.info("Stopping Kafka Producer for topic: {}", myTopic);
>> if (myProducer != null) {
>> myProducer.close();
>> }
>> }
>>
>>
>>
>> On Fri, Jan 30, 2015 at 11:22 AM, Manikumar Reddy <ku...@nmsworks.co.in>
>> wrote:
>>
>> Hope you are closing the producers. can you share the attachment through
>>> gist/patebin
>>>
>>> On Fri, Jan 30, 2015 at 11:11 AM, ankit tyagi <
>>> ankittyagi.mnnit@gmail.com>
>>> wrote:
>>>
>>> Hi Jaikiran,
>>>>
>>>> I am using ubuntu and was able to reproduce on redhat too. Please find
>>>>
>>> the
>>>
>>>> more information below.
>>>>
>>>>
>>>> *DISTRIB_ID=Ubuntu*
>>>> *DISTRIB_RELEASE=12.04*
>>>> *DISTRIB_CODENAME=precise*
>>>> *DISTRIB_DESCRIPTION="Ubuntu 12.04.5 LTS"*
>>>>
>>>> *java version "1.7.0_72"*
>>>>
>>>> This is happening on client side. Output of lsof was showing that
>>>> maximum
>>>> fd were FIFO and anon. But after GC FD count was reduced significantly.
>>>>
>>>> Below is my Client Code which i am using for publishing message.
>>>>
>>>>
>>>> * private Producer<KafkaPartitionKey, KafkaEventWrapper> myProducer;*
>>>>
>>>> * myProducer = new Producer<>(new
>>>> ProducerConfig(myProducerProperties));*
>>>>
>>>> * public void send(*
>>>> * List<KeyedMessage<KafkaPartitionKey, KafkaEventWrapper>>
>>>> msgs)*
>>>> * {*
>>>> * myProducer.send(msgs);*
>>>> * }*
>>>>
>>>>
>>>> we are using sync producer. I am attaching object histo before
>>>>
>>> GC(histo_1)
>>>
>>>> and after GC(histo_2) in my application.
>>>>
>>>> On Fri, Jan 30, 2015 at 9:34 AM, Jaikiran Pai <jai.forums2013@gmail.com
>>>> >
>>>> wrote:
>>>>
>>>> Which operating system are you on and what Java version? Depending on
>>>>>
>>>> the
>>>
>>>> OS, you could get tools (like lsof) to show which file descriptors are
>>>>> being held on to. Is it the client JVM which ends up with these leaks?
>>>>>
>>>>> Also, would it be possible to post a snippet of your application code
>>>>> which shows how you are using the Kafka APIs?
>>>>>
>>>>> -Jaikiran
>>>>> On Thursday 29 January 2015 04:36 PM, ankit tyagi wrote:
>>>>>
>>>>> Hi,
>>>>>>
>>>>>> Currently we are using sync producer client of 0.8.1 version in our
>>>>>> production box . we are getting the following exception while
>>>>>>
>>>>> publishing
>>>
>>>> kafka message
>>>>>>
>>>>>> *[2015-01-29
>>>>>> 13:21:45.505][ThreadPoolTaskExecutor-603][WARN][ClientUtils$:89]
>>>>>>
>>>>> Fetching
>>>>
>>>>> topic metadata with correlation id 10808 for topics [Set(*
>>>>>> *kafka_topic_coms_FD_test1)] from broker
>>>>>>
>>>>> [id:0,host:localhost,port:9092]
>>>
>>>> failed*
>>>>>> *java.net.ConnectException: Connection refused*
>>>>>> * at sun.nio.ch.Net.connect0(Native Method)*
>>>>>> * at sun.nio.ch.Net.connect(Net.java:465)*
>>>>>> * at sun.nio.ch.Net.connect(Net.java:457)*
>>>>>> * at
>>>>>> sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:670)*
>>>>>> at
>>>>>>
>>>>> kafka.network.BlockingChannel.connect(BlockingChannel.scala:
>>>
>>>> 57)
>>>>>> at
>>>>>>
>>>>> kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
>>>
>>>> at
>>>>>>
>>>>>> kafka.producer.SyncProducer.getOrMakeConnection(
>>> SyncProducer.scala:156)
>>>
>>>> at
>>>>>> kafka.producer.SyncProducer.kafka$producer$SyncProducer$$
>>>>>> doSend(SyncProducer.scala:68)
>>>>>> at kafka.producer.SyncProducer.send(SyncProducer.scala:112)
>>>>>> at
>>>>>> kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
>>>>>> at
>>>>>> kafka.producer.BrokerPartitionInfo.updateInfo(
>>>>>> BrokerPartitionInfo.scala:82)
>>>>>>
>>>>>>
>>>>>> we are using dynamic thread pool to publish message to kafka. My
>>>>>> observation is when after keep alive time when threads in my executor
>>>>>>
>>>>> gets
>>>>
>>>>> destroyed, somehow file descriptor is not getting cleared but when i
>>>>>>
>>>>> did
>>>
>>>> explicitly ran the full gc, fd count got reduced by a signification
>>>>>>
>>>>> amout.
>>>>
>>>>>
>>>>>>
>
Re: Potential socket leak in kafka sync producer
Posted by Jaikiran Pai <ja...@gmail.com>.
What kind of a (managed) component is that which has the @PreDestroy?
Looking at the previous snippet you added, it looks like you are
creating the Producer in some method? If you are going to close the
producer in a @PreDestroy of the component, then you should be creating
the producer in the @PostConstruct of the same component, so that you
have proper lifecycle management of those resources.
-Jaikiran
On Friday 30 January 2015 12:20 PM, ankit tyagi wrote:
> Hi,
>
> I am closing my producer at the time of shutting down my application.
>
> @PreDestroy
> public void stop()
> {
> LOG.info("Stopping Kafka Producer for topic: {}", myTopic);
> if (myProducer != null) {
> myProducer.close();
> }
> }
>
>
>
> On Fri, Jan 30, 2015 at 11:22 AM, Manikumar Reddy <ku...@nmsworks.co.in>
> wrote:
>
>> Hope you are closing the producers. can you share the attachment through
>> gist/patebin
>>
>> On Fri, Jan 30, 2015 at 11:11 AM, ankit tyagi <an...@gmail.com>
>> wrote:
>>
>>> Hi Jaikiran,
>>>
>>> I am using ubuntu and was able to reproduce on redhat too. Please find
>> the
>>> more information below.
>>>
>>>
>>> *DISTRIB_ID=Ubuntu*
>>> *DISTRIB_RELEASE=12.04*
>>> *DISTRIB_CODENAME=precise*
>>> *DISTRIB_DESCRIPTION="Ubuntu 12.04.5 LTS"*
>>>
>>> *java version "1.7.0_72"*
>>>
>>> This is happening on client side. Output of lsof was showing that maximum
>>> fd were FIFO and anon. But after GC FD count was reduced significantly.
>>>
>>> Below is my Client Code which i am using for publishing message.
>>>
>>>
>>> * private Producer<KafkaPartitionKey, KafkaEventWrapper> myProducer;*
>>>
>>> * myProducer = new Producer<>(new
>>> ProducerConfig(myProducerProperties));*
>>>
>>> * public void send(*
>>> * List<KeyedMessage<KafkaPartitionKey, KafkaEventWrapper>> msgs)*
>>> * {*
>>> * myProducer.send(msgs);*
>>> * }*
>>>
>>>
>>> we are using sync producer. I am attaching object histo before
>> GC(histo_1)
>>> and after GC(histo_2) in my application.
>>>
>>> On Fri, Jan 30, 2015 at 9:34 AM, Jaikiran Pai <ja...@gmail.com>
>>> wrote:
>>>
>>>> Which operating system are you on and what Java version? Depending on
>> the
>>>> OS, you could get tools (like lsof) to show which file descriptors are
>>>> being held on to. Is it the client JVM which ends up with these leaks?
>>>>
>>>> Also, would it be possible to post a snippet of your application code
>>>> which shows how you are using the Kafka APIs?
>>>>
>>>> -Jaikiran
>>>> On Thursday 29 January 2015 04:36 PM, ankit tyagi wrote:
>>>>
>>>>> Hi,
>>>>>
>>>>> Currently we are using sync producer client of 0.8.1 version in our
>>>>> production box . we are getting the following exception while
>> publishing
>>>>> kafka message
>>>>>
>>>>> *[2015-01-29
>>>>> 13:21:45.505][ThreadPoolTaskExecutor-603][WARN][ClientUtils$:89]
>>> Fetching
>>>>> topic metadata with correlation id 10808 for topics [Set(*
>>>>> *kafka_topic_coms_FD_test1)] from broker
>> [id:0,host:localhost,port:9092]
>>>>> failed*
>>>>> *java.net.ConnectException: Connection refused*
>>>>> * at sun.nio.ch.Net.connect0(Native Method)*
>>>>> * at sun.nio.ch.Net.connect(Net.java:465)*
>>>>> * at sun.nio.ch.Net.connect(Net.java:457)*
>>>>> * at
>>>>> sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:670)*
>>>>> at
>> kafka.network.BlockingChannel.connect(BlockingChannel.scala:
>>>>> 57)
>>>>> at
>> kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
>>>>> at
>>>>>
>> kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:156)
>>>>> at
>>>>> kafka.producer.SyncProducer.kafka$producer$SyncProducer$$
>>>>> doSend(SyncProducer.scala:68)
>>>>> at kafka.producer.SyncProducer.send(SyncProducer.scala:112)
>>>>> at
>>>>> kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
>>>>> at
>>>>> kafka.producer.BrokerPartitionInfo.updateInfo(
>>>>> BrokerPartitionInfo.scala:82)
>>>>>
>>>>>
>>>>> we are using dynamic thread pool to publish message to kafka. My
>>>>> observation is when after keep alive time when threads in my executor
>>> gets
>>>>> destroyed, somehow file descriptor is not getting cleared but when i
>> did
>>>>> explicitly ran the full gc, fd count got reduced by a signification
>>> amout.
>>>>>
Re: Potential socket leak in kafka sync producer
Posted by ankit tyagi <an...@gmail.com>.
Hi,
I am closing my producer at the time of shutting down my application.
@PreDestroy
public void stop()
{
LOG.info("Stopping Kafka Producer for topic: {}", myTopic);
if (myProducer != null) {
myProducer.close();
}
}
On Fri, Jan 30, 2015 at 11:22 AM, Manikumar Reddy <ku...@nmsworks.co.in>
wrote:
> Hope you are closing the producers. can you share the attachment through
> gist/patebin
>
> On Fri, Jan 30, 2015 at 11:11 AM, ankit tyagi <an...@gmail.com>
> wrote:
>
> > Hi Jaikiran,
> >
> > I am using ubuntu and was able to reproduce on redhat too. Please find
> the
> > more information below.
> >
> >
> > *DISTRIB_ID=Ubuntu*
> > *DISTRIB_RELEASE=12.04*
> > *DISTRIB_CODENAME=precise*
> > *DISTRIB_DESCRIPTION="Ubuntu 12.04.5 LTS"*
> >
> > *java version "1.7.0_72"*
> >
> > This is happening on client side. Output of lsof was showing that maximum
> > fd were FIFO and anon. But after GC FD count was reduced significantly.
> >
> > Below is my Client Code which i am using for publishing message.
> >
> >
> > * private Producer<KafkaPartitionKey, KafkaEventWrapper> myProducer;*
> >
> > * myProducer = new Producer<>(new
> > ProducerConfig(myProducerProperties));*
> >
> > * public void send(*
> > * List<KeyedMessage<KafkaPartitionKey, KafkaEventWrapper>> msgs)*
> > * {*
> > * myProducer.send(msgs);*
> > * }*
> >
> >
> > we are using sync producer. I am attaching object histo before
> GC(histo_1)
> > and after GC(histo_2) in my application.
> >
> > On Fri, Jan 30, 2015 at 9:34 AM, Jaikiran Pai <ja...@gmail.com>
> > wrote:
> >
> > >
> > > Which operating system are you on and what Java version? Depending on
> the
> > > OS, you could get tools (like lsof) to show which file descriptors are
> > > being held on to. Is it the client JVM which ends up with these leaks?
> > >
> > > Also, would it be possible to post a snippet of your application code
> > > which shows how you are using the Kafka APIs?
> > >
> > > -Jaikiran
> > > On Thursday 29 January 2015 04:36 PM, ankit tyagi wrote:
> > >
> > >> Hi,
> > >>
> > >> Currently we are using sync producer client of 0.8.1 version in our
> > >> production box . we are getting the following exception while
> publishing
> > >> kafka message
> > >>
> > >> *[2015-01-29
> > >> 13:21:45.505][ThreadPoolTaskExecutor-603][WARN][ClientUtils$:89]
> > Fetching
> > >> topic metadata with correlation id 10808 for topics [Set(*
> > >> *kafka_topic_coms_FD_test1)] from broker
> [id:0,host:localhost,port:9092]
> > >> failed*
> > >> *java.net.ConnectException: Connection refused*
> > >> * at sun.nio.ch.Net.connect0(Native Method)*
> > >> * at sun.nio.ch.Net.connect(Net.java:465)*
> > >> * at sun.nio.ch.Net.connect(Net.java:457)*
> > >> * at
> > >> sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:670)*
> > >> at
> kafka.network.BlockingChannel.connect(BlockingChannel.scala:
> > >> 57)
> > >> at
> kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
> > >> at
> > >>
> kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:156)
> > >> at
> > >> kafka.producer.SyncProducer.kafka$producer$SyncProducer$$
> > >> doSend(SyncProducer.scala:68)
> > >> at kafka.producer.SyncProducer.send(SyncProducer.scala:112)
> > >> at
> > >> kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
> > >> at
> > >> kafka.producer.BrokerPartitionInfo.updateInfo(
> > >> BrokerPartitionInfo.scala:82)
> > >>
> > >>
> > >> we are using dynamic thread pool to publish message to kafka. My
> > >> observation is when after keep alive time when threads in my executor
> > gets
> > >> destroyed, somehow file descriptor is not getting cleared but when i
> did
> > >> explicitly ran the full gc, fd count got reduced by a signification
> > amout.
> > >>
> > >>
> > >
> >
>
Re: Potential socket leak in kafka sync producer
Posted by Manikumar Reddy <ku...@nmsworks.co.in>.
Hope you are closing the producers. can you share the attachment through
gist/patebin
On Fri, Jan 30, 2015 at 11:11 AM, ankit tyagi <an...@gmail.com>
wrote:
> Hi Jaikiran,
>
> I am using ubuntu and was able to reproduce on redhat too. Please find the
> more information below.
>
>
> *DISTRIB_ID=Ubuntu*
> *DISTRIB_RELEASE=12.04*
> *DISTRIB_CODENAME=precise*
> *DISTRIB_DESCRIPTION="Ubuntu 12.04.5 LTS"*
>
> *java version "1.7.0_72"*
>
> This is happening on client side. Output of lsof was showing that maximum
> fd were FIFO and anon. But after GC FD count was reduced significantly.
>
> Below is my Client Code which i am using for publishing message.
>
>
> * private Producer<KafkaPartitionKey, KafkaEventWrapper> myProducer;*
>
> * myProducer = new Producer<>(new
> ProducerConfig(myProducerProperties));*
>
> * public void send(*
> * List<KeyedMessage<KafkaPartitionKey, KafkaEventWrapper>> msgs)*
> * {*
> * myProducer.send(msgs);*
> * }*
>
>
> we are using sync producer. I am attaching object histo before GC(histo_1)
> and after GC(histo_2) in my application.
>
> On Fri, Jan 30, 2015 at 9:34 AM, Jaikiran Pai <ja...@gmail.com>
> wrote:
>
> >
> > Which operating system are you on and what Java version? Depending on the
> > OS, you could get tools (like lsof) to show which file descriptors are
> > being held on to. Is it the client JVM which ends up with these leaks?
> >
> > Also, would it be possible to post a snippet of your application code
> > which shows how you are using the Kafka APIs?
> >
> > -Jaikiran
> > On Thursday 29 January 2015 04:36 PM, ankit tyagi wrote:
> >
> >> Hi,
> >>
> >> Currently we are using sync producer client of 0.8.1 version in our
> >> production box . we are getting the following exception while publishing
> >> kafka message
> >>
> >> *[2015-01-29
> >> 13:21:45.505][ThreadPoolTaskExecutor-603][WARN][ClientUtils$:89]
> Fetching
> >> topic metadata with correlation id 10808 for topics [Set(*
> >> *kafka_topic_coms_FD_test1)] from broker [id:0,host:localhost,port:9092]
> >> failed*
> >> *java.net.ConnectException: Connection refused*
> >> * at sun.nio.ch.Net.connect0(Native Method)*
> >> * at sun.nio.ch.Net.connect(Net.java:465)*
> >> * at sun.nio.ch.Net.connect(Net.java:457)*
> >> * at
> >> sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:670)*
> >> at kafka.network.BlockingChannel.connect(BlockingChannel.scala:
> >> 57)
> >> at kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
> >> at
> >> kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:156)
> >> at
> >> kafka.producer.SyncProducer.kafka$producer$SyncProducer$$
> >> doSend(SyncProducer.scala:68)
> >> at kafka.producer.SyncProducer.send(SyncProducer.scala:112)
> >> at
> >> kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
> >> at
> >> kafka.producer.BrokerPartitionInfo.updateInfo(
> >> BrokerPartitionInfo.scala:82)
> >>
> >>
> >> we are using dynamic thread pool to publish message to kafka. My
> >> observation is when after keep alive time when threads in my executor
> gets
> >> destroyed, somehow file descriptor is not getting cleared but when i did
> >> explicitly ran the full gc, fd count got reduced by a signification
> amout.
> >>
> >>
> >
>
Re: Potential socket leak in kafka sync producer
Posted by ankit tyagi <an...@gmail.com>.
Hi Jaikiran,
I am using ubuntu and was able to reproduce on redhat too. Please find the
more information below.
*DISTRIB_ID=Ubuntu*
*DISTRIB_RELEASE=12.04*
*DISTRIB_CODENAME=precise*
*DISTRIB_DESCRIPTION="Ubuntu 12.04.5 LTS"*
*java version "1.7.0_72"*
This is happening on client side. Output of lsof was showing that maximum
fd were FIFO and anon. But after GC FD count was reduced significantly.
Below is my Client Code which i am using for publishing message.
* private Producer<KafkaPartitionKey, KafkaEventWrapper> myProducer;*
* myProducer = new Producer<>(new
ProducerConfig(myProducerProperties));*
* public void send(*
* List<KeyedMessage<KafkaPartitionKey, KafkaEventWrapper>> msgs)*
* {*
* myProducer.send(msgs);*
* }*
we are using sync producer. I am attaching object histo before GC(histo_1)
and after GC(histo_2) in my application.
On Fri, Jan 30, 2015 at 9:34 AM, Jaikiran Pai <ja...@gmail.com>
wrote:
>
> Which operating system are you on and what Java version? Depending on the
> OS, you could get tools (like lsof) to show which file descriptors are
> being held on to. Is it the client JVM which ends up with these leaks?
>
> Also, would it be possible to post a snippet of your application code
> which shows how you are using the Kafka APIs?
>
> -Jaikiran
> On Thursday 29 January 2015 04:36 PM, ankit tyagi wrote:
>
>> Hi,
>>
>> Currently we are using sync producer client of 0.8.1 version in our
>> production box . we are getting the following exception while publishing
>> kafka message
>>
>> *[2015-01-29
>> 13:21:45.505][ThreadPoolTaskExecutor-603][WARN][ClientUtils$:89] Fetching
>> topic metadata with correlation id 10808 for topics [Set(*
>> *kafka_topic_coms_FD_test1)] from broker [id:0,host:localhost,port:9092]
>> failed*
>> *java.net.ConnectException: Connection refused*
>> * at sun.nio.ch.Net.connect0(Native Method)*
>> * at sun.nio.ch.Net.connect(Net.java:465)*
>> * at sun.nio.ch.Net.connect(Net.java:457)*
>> * at
>> sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:670)*
>> at kafka.network.BlockingChannel.connect(BlockingChannel.scala:
>> 57)
>> at kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
>> at
>> kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:156)
>> at
>> kafka.producer.SyncProducer.kafka$producer$SyncProducer$$
>> doSend(SyncProducer.scala:68)
>> at kafka.producer.SyncProducer.send(SyncProducer.scala:112)
>> at
>> kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
>> at
>> kafka.producer.BrokerPartitionInfo.updateInfo(
>> BrokerPartitionInfo.scala:82)
>>
>>
>> we are using dynamic thread pool to publish message to kafka. My
>> observation is when after keep alive time when threads in my executor gets
>> destroyed, somehow file descriptor is not getting cleared but when i did
>> explicitly ran the full gc, fd count got reduced by a signification amout.
>>
>>
>
Re: Potential socket leak in kafka sync producer
Posted by Jaikiran Pai <ja...@gmail.com>.
Which operating system are you on and what Java version? Depending on
the OS, you could get tools (like lsof) to show which file descriptors
are being held on to. Is it the client JVM which ends up with these leaks?
Also, would it be possible to post a snippet of your application code
which shows how you are using the Kafka APIs?
-Jaikiran
On Thursday 29 January 2015 04:36 PM, ankit tyagi wrote:
> Hi,
>
> Currently we are using sync producer client of 0.8.1 version in our
> production box . we are getting the following exception while publishing
> kafka message
>
> *[2015-01-29
> 13:21:45.505][ThreadPoolTaskExecutor-603][WARN][ClientUtils$:89] Fetching
> topic metadata with correlation id 10808 for topics [Set(*
> *kafka_topic_coms_FD_test1)] from broker [id:0,host:localhost,port:9092]
> failed*
> *java.net.ConnectException: Connection refused*
> * at sun.nio.ch.Net.connect0(Native Method)*
> * at sun.nio.ch.Net.connect(Net.java:465)*
> * at sun.nio.ch.Net.connect(Net.java:457)*
> * at
> sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:670)*
> at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57)
> at kafka.producer.SyncProducer.connect(SyncProducer.scala:141)
> at
> kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:156)
> at
> kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68)
> at kafka.producer.SyncProducer.send(SyncProducer.scala:112)
> at
> kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
> at
> kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
>
>
> we are using dynamic thread pool to publish message to kafka. My
> observation is when after keep alive time when threads in my executor gets
> destroyed, somehow file descriptor is not getting cleared but when i did
> explicitly ran the full gc, fd count got reduced by a signification amout.
>