Skip to content
This repository was archived by the owner on Jul 15, 2023. It is now read-only.

Fix for RecreateSyncProducerPoolForMetadata #34

Open
Maxime Caron (maximecaron) wants to merge 5 commits into
base: master
Choose a base branch
from
Open

Fix for RecreateSyncProducerPoolForMetadata #34

Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e4ed84a
call RecreateSyncProducerPoolForMetadata() in RefreshMetadata retry loop
maximecaron Mar 17, 2016
dc9c4d1
add continue; to not increment retry twice
maximecaron Mar 17, 2016
9b3ad1f
Fix build error
maximecaron Mar 17, 2016
31c682f
Remove retry loop inside KafkaSimpleManager.RefreshMetadata
maximecaron May 18, 2016
e44b973
Update KafkaSimpleManager.cs
maximecaron May 18, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 31 additions & 16 deletions src/KafkaNET.Library/Helper/KafkaSimpleManager.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ namespace Kafka.Client.Helper
using Kafka.Client.Producers.Sync;
using Kafka.Client.Requests;
using Kafka.Client.Utils;
using Microsoft.KafkaNET.Library.Util;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
Expand Down Expand Up @@ -177,31 +178,46 @@ public TopicMetadata RefreshMetadata(short versionId, string clientId, int corre
Logger.InfoFormat("RefreshMetadata enter: {0} {1} {2} Topic:{3} Force:{4}", versionId, clientId, correlationId, topic, force);
if (!force && this.TopicMetadatas.ContainsKey(topic))
return this.TopicMetadatas[topic];

int maxRetryCount = 2;
int retry = 0;
while (retry < 2)
while (retry < maxRetryCount)
{
Dictionary<string, TopicMetadata> tempTopicMetadatas = new Dictionary<string, TopicMetadata>();
Dictionary<string, DateTime> tempTopicMetadatasLastUpdateTime = new Dictionary<string, DateTime>();
Dictionary<int, Tuple<Broker, BrokerConfiguration>> partitionLeaders = new Dictionary<int, Tuple<Broker, BrokerConfiguration>>();
RefreshMetadataInternal(versionId, clientId, correlationId, topic, tempTopicMetadatas, tempTopicMetadatasLastUpdateTime, partitionLeaders);

if (tempTopicMetadatas.ContainsKey(topic))
try
{
this.TopicMetadatas[topic] = tempTopicMetadatas[topic];
this.TopicMetadatasLastUpdateTime[topic] = tempTopicMetadatasLastUpdateTime[topic];
this.TopicMetadataPartitionsLeaders[topic] = partitionLeaders;
int partitionCountInZK = GetTopicPartitionsFromZK(topic).Count;
if (partitionCountInZK != partitionLeaders.Count)
Logger.WarnFormat("RefreshMetadata exit return. Some partitions has no leader. Topic:{0} PartitionMetadata:{1} partitionLeaders:{2} != partitionCountInZK:{3}", topic, tempTopicMetadatas[topic].PartitionsMetadata.Count(), partitionLeaders.Count, partitionCountInZK);
RefreshMetadataInternal(versionId, clientId, correlationId, topic, tempTopicMetadatas, tempTopicMetadatasLastUpdateTime, partitionLeaders);

if (tempTopicMetadatas.ContainsKey(topic))
{
this.TopicMetadatas[topic] = tempTopicMetadatas[topic];
this.TopicMetadatasLastUpdateTime[topic] = tempTopicMetadatasLastUpdateTime[topic];
this.TopicMetadataPartitionsLeaders[topic] = partitionLeaders;
int partitionCountInZK = GetTopicPartitionsFromZK(topic).Count;
if (partitionCountInZK != partitionLeaders.Count)
Logger.WarnFormat("RefreshMetadata exit return. Some partitions has no leader. Topic:{0} PartitionMetadata:{1} partitionLeaders:{2} != partitionCountInZK:{3}", topic, tempTopicMetadatas[topic].PartitionsMetadata.Count(), partitionLeaders.Count, partitionCountInZK);
else
Logger.InfoFormat("RefreshMetadata exit return. Topic:{0} PartitionMetadata:{1} partitionLeaders:{2} partitionCountInZK:{3}", topic, tempTopicMetadatas[topic].PartitionsMetadata.Count(), partitionLeaders.Count, partitionCountInZK);
return this.TopicMetadatas[topic];
}
else
Logger.InfoFormat("RefreshMetadata exit return. Topic:{0} PartitionMetadata:{1} partitionLeaders:{2} partitionCountInZK:{3}", topic, tempTopicMetadatas[topic].PartitionsMetadata.Count(), partitionLeaders.Count, partitionCountInZK);
return this.TopicMetadatas[topic];
{
Logger.WarnFormat("Got null for metadata of topic {0}, will RecreateSyncProducerPoolForMetadata and retry . ", topic);
RecreateSyncProducerPoolForMetadata();
}
}
else
catch (Exception ex)
{
Logger.WarnFormat("Got null for metadata of topic {0}, will RecreateSyncProducerPoolForMetadata and retry . ", topic);
Logger.WarnFormat("Got exception while refreshing metadata of topic {0}, will RecreateSyncProducerPoolForMetadata and retry . {1} ",topic,
ExceptionUtil.GetExceptionDetailInfo(ex));
RecreateSyncProducerPoolForMetadata();

@mhorstma mhorstma Apr 6, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Only concern is that RecreateSyncProducerPoolForMetadata() can throw. I hate to see new exceptions thrown from inside a catch block. Could this be restructured to move this call be into its own try/catch or just extracted from the current try/catch altogether and let it throw if its going to? If it does, it probably indicates a more severe error where a retry may not help.

@maximecaron Maxime Caron (maximecaron) Apr 6, 2016

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I dont think RecreateSyncProducerPoolForMetadata() should be extracted/moved outside the catch block as it should not need to be called each time RefreshMetadata() is called but only if RefreshMetadata() fail.

If what you mean is something such as setting a bool flag inside the catch block to indicate a failure and then call RecreateSyncProducerPoolForMetadata() outside of the catch block if the flag was set, this could be done and would have the same intended result. but I am not sure I understand how this will make it better.

We could also add an extra try/catch around RecreateSyncProducerPoolForMetadata(); to honor the retryCount if you think this is better, but the retry loop was not really needed to start with since the caller of RefreshMetadata() could simply retry calling RefreshMetadata() if an exception is thrown.

@mhorstma mhorstma Apr 6, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There are two issues here, really. One is that the stack trace for any exceptions thrown from within the catch block will be convoluted (not the end of the world), and two is that this code is using the exception for flow control. IF the code in the try block throws an exception, THEN execute RecreateSyncProducerPoolForMetadata(). That part makes me most uncomfortable. Could you tighten up the try block to only wrap RefreshMetadataInternal, the problem method, then set a bool in the catch block and if the bool is true, call Recreate...() lower down? That way any exceptions are handled and execution moves on to an explicit flow control block.

Sorry to push on this. This is a needed fix, but this lib is already pretty liberal with exceptions and being more disciplined about how they are used in new code seems like a good thing.

@maximecaron Maxime Caron (maximecaron) Apr 6, 2016

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sure wrapping only RefreshMetadataInternal and setting bool sounds good to me.

retry++;
if (retry >= maxRetryCount)
{
throw ex;
}
continue;
}
retry++;
}
Expand Down Expand Up @@ -241,7 +257,6 @@ internal BrokerConfiguration GetLeaderBrokerOfPartition(string topic, int partit
private void RefreshMetadataInternal(short versionId, string clientId, int correlationId, string topic, Dictionary<string, TopicMetadata> tempTopicMetadatas, Dictionary<string, DateTime> tempTopicMetadatasLastUpdateTime, Dictionary<int, Tuple<Broker, BrokerConfiguration>> partitionLeaders)
{
Logger.InfoFormat("RefreshMetadataInternal enter: {0} {1} {2} Topic:{3} ", versionId, clientId, correlationId, topic);

lock (syncProducerPoolForMetadataLock)
{
BrokerPartitionInfo brokerPartitionInfo = new BrokerPartitionInfo(this.syncProducerPoolForMetaData, tempTopicMetadatas, tempTopicMetadatasLastUpdateTime, ProducerConfiguration.DefaultTopicMetaDataRefreshIntervalMS, this.syncProducerPoolForMetaData.zkClient);
Expand Down