AndrewJSchofield commented on code in PR #23964:
URL: https://github.com/apache/kafka/pull/23964#discussion_r4380601609
##########
server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java:
##########
Review Comment:
This comment is misleading now I think.
##########
core/src/main/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImpl.java:
##########
@@ -111,17 +119,23 @@ public boolean isDlqEnabledOnTopic(String topic) {
@Override
public int dlqTopicMaxMessageBytes(String topic) {
Properties props = metadataCache.topicConfig(topic);
- // LogConfig defines its own static default for this key, so an
explicit containsKey
- // check is needed to distinguish "no topic-level override" from
"override present"
- // and fall back to the broker's configured message.max.bytes in the
former case.
- if (props == null ||
!props.containsKey(TopicConfig.MAX_MESSAGE_BYTES_CONFIG)) {
+ // LogConfig defines its own static default for this key, so only a
topic-level override is read
+ // here and without one the broker's configured message.max.bytes is
used. Reading the one key
+ // directly avoids parsing every topic config, as constructing a
LogConfig would.
Review Comment:
Ditto about LogConfig.
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -178,6 +178,9 @@ class BrokerServer(
private var shareGroupDLQManager: ShareGroupDLQManager = _
+ // Runs the source record reads of the share group DLQ manager, off the
persister thread.
+ private var shareGroupDLQFetchExecutor: ExecutorService = _
Review Comment:
I wonder whether the ExecutorService ought to be within the DLQ manager.
##########
core/src/main/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImpl.java:
##########
@@ -101,8 +102,15 @@ public boolean isDlqEnabledOnTopic(String topic) {
if (props == null) {
return false;
}
+ // This is called for every DLQ request, so read the one flag
directly. Constructing a LogConfig
Review Comment:
The "Constructing a LogConfig...." sentence is justification for the PR,
rather than a useful thing to have in a comment.
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -774,17 +777,28 @@ class BrokerServer(
}
}
+ /** Daemon thread factory naming each thread with the given name. */
+ private def daemonThreadFactory(threadName: String): ThreadFactory = {
+ (runnable: Runnable) => {
+ val thread = new Thread(runnable, threadName)
+ thread.setDaemon(true)
+ thread
+ }
+ }
+
private def createShareGroupDLQManager(): ShareGroupDLQManager = {
val className = config.shareGroupConfig.shareGroupDLQManagerClassName
if (className.nonEmpty) {
if (className.equals(classOf[DefaultShareGroupDLQManager].getName)) {
+ shareGroupDLQFetchExecutor =
Executors.newSingleThreadExecutor(daemonThreadFactory("share-group-dlq-record-fetcher"))
Review Comment:
Using a daemon thread is an improvement. Would a small number of threads be
better than just one?
##########
core/src/main/java/kafka/server/share/ShareCoordinatorMetadataCacheHelperImpl.java:
##########
@@ -111,17 +119,23 @@ public boolean isDlqEnabledOnTopic(String topic) {
@Override
public int dlqTopicMaxMessageBytes(String topic) {
Properties props = metadataCache.topicConfig(topic);
- // LogConfig defines its own static default for this key, so an
explicit containsKey
- // check is needed to distinguish "no topic-level override" from
"override present"
- // and fall back to the broker's configured message.max.bytes in the
former case.
- if (props == null ||
!props.containsKey(TopicConfig.MAX_MESSAGE_BYTES_CONFIG)) {
+ // LogConfig defines its own static default for this key, so only a
topic-level override is read
+ // here and without one the broker's configured message.max.bytes is
used. Reading the one key
+ // directly avoids parsing every topic config, as constructing a
LogConfig would.
+ Object value = props == null ? null :
props.get(TopicConfig.MAX_MESSAGE_BYTES_CONFIG);
+ if (value == null) {
return messageMaxBytesSupplier.getAsInt();
}
try {
- return new
LogConfig(props).getInt(TopicConfig.MAX_MESSAGE_BYTES_CONFIG);
+ Integer maxMessageBytes = (Integer)
ConfigDef.parseType(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, value,
ConfigDef.Type.INT);
+ // LogConfig requires message.max.bytes to be at least 0.
Review Comment:
Won't that already be true because of the configuration validation for the
topic?
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -943,6 +957,15 @@ class BrokerServer(
if (shareGroupDLQManager != null)
Utils.swallow(this.logger.underlying, () =>
shareGroupDLQManager.stop())
+ // Stopped after the DLQ manager so that nothing submits to it any more.
Reads still running or
+ // queued are given a moment to finish and are interrupted otherwise.
+ if (shareGroupDLQFetchExecutor != null)
+ Utils.swallow(this.logger.underlying, () => {
+ shareGroupDLQFetchExecutor.shutdown()
Review Comment:
Doesn't this use ReplicaManager? Is the shutdown order sensible?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
