Skip to content

KAFKA-15853: Move TransactionLogConfig and TransactionStateManagerConfig getters out of KafkaConfig #16665

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chia7712 merged 8 commits into from
Sep 3, 2024
Merged

KAFKA-15853: Move TransactionLogConfig and TransactionStateManagerConfig getters out of KafkaConfig #16665

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
d643e8c
KAFKA-15853: Refactor TransactionLogConfig and TransactionStateManag…
OmniaGM Jul 28, 2024
fef8123
Merge remote-tracking branch 'apache' into KAFKA-15853-txnConfig
OmniaGM Jul 29, 2024
5a4d3d5
address @chia7712's feedback
OmniaGM Aug 1, 2024
2c57877
Merge remote-tracking branch 'apache/trunk' into KAFKA-15853-txnConfig
OmniaGM Aug 13, 2024
801dca0
Merge remote-tracking branch 'apache/trunk' into KAFKA-15853-txnConfig
OmniaGM Aug 13, 2024
c754ca1
Merge remote-tracking branch 'apache/trunk' into KAFKA-15853-txnConfig
OmniaGM Aug 19, 2024
fbedd41
Merge remote-tracking branch 'apache/trunk' into KAFKA-15853-txnConfig
OmniaGM Aug 21, 2024
1396b7d
Merge remote-tracking branch 'apache/trunk' into KAFKA-15853-txnConfig
OmniaGM Sep 2, 2024
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
Prev Previous commit
Next Next commit
address @chia7712's feedback
  • Loading branch information
@OmniaGM
OmniaGM committed Aug 1, 2024
commit 5a4d3d5d0dfbe54a0a71a22a7913370447077bf7
26 changes: 13 additions & 13 deletions ...rdinator/src/main/java/org/apache/kafka/coordinator/transaction/TransactionLogConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,13 +74,13 @@ public final class TransactionLogConfig {
// Configuration for testing only as default value should be sufficient for typical usage
.defineInternal(PRODUCER_ID_EXPIRATION_CHECK_INTERVAL_MS_CONFIG, INT, PRODUCER_ID_EXPIRATION_CHECK_INTERVAL_MS_DEFAULT, atLeast(1), LOW, PRODUCER_ID_EXPIRATION_CHECK_INTERVAL_MS_DOC);

private AbstractConfig config;
private int transactionTopicMinISR;
private int transactionsLoadBufferSize;
private short transactionTopicReplicationFactor;
private int transactionTopicPartitions;
private int transactionTopicSegmentBytes;
private int producerIdExpirationCheckIntervalMs;
private final AbstractConfig config;
private final int transactionTopicMinISR;
private final int transactionsLoadBufferSize;
private final short transactionTopicReplicationFactor;
private final int transactionTopicPartitions;
private final int transactionTopicSegmentBytes;
private final int producerIdExpirationCheckIntervalMs;

public TransactionLogConfig(AbstractConfig config) {
this.config = config;
Expand Down Expand Up @@ -112,17 +112,17 @@ public int transactionTopicSegmentBytes() {
return transactionTopicSegmentBytes;
}

public int producerIdExpirationCheckIntervalMs() {
return producerIdExpirationCheckIntervalMs;
}

// This is a broker dynamic config used for DynamicProducerStateManagerConfig
public Boolean transactionPartitionVerificationEnable() {
return config.getBoolean(TRANSACTION_PARTITION_VERIFICATION_ENABLE_CONFIG);
Copy link
Copy Markdown
Member

@chia7712 chia7712 Jul 30, 2024

Choose a reason for hiding this comment

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

could you please add comments to for those dynamic configs?

Copy link
Copy Markdown
Member

@mimaison mimaison Aug 8, 2024

Choose a reason for hiding this comment

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

Even with the comments, I'm not sure I understand why this is not initialized in the constructor.

Can you explain how this is supposed to work? AbstractConfig instances are kind of read only by default but it looks like we're expecting it to change here. Am I missing something?

Copy link
Copy Markdown
Member

@chia7712 chia7712 Aug 8, 2024

Choose a reason for hiding this comment

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

we had a discussion for that before (#16458 (comment))

AbstractConfig is read-only, but the sub class KafkaConfig can update inner configs dynamically (see https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/KafkaConfig.scala#L192). Hence, the passed AbstractConfig always have "up-to-date configs"

There is a jira which tries to refactor it (https://issues.apache.org/jira/browse/KAFKA-17001), but I haven't worked on it because that will introduce major changes to scala code.

In short, the basic rules for now are shown below.

  1. the non-dynamic config will be evaluated in constructor.
  2. the dynamic config will be evaluated in getter (as AbstractConfig always have latest configs)

Copy link
Copy Markdown
Member

@mimaison mimaison Aug 12, 2024

Choose a reason for hiding this comment

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

Thanks for the explanations!

}

// This is a broker dynamic config used for DynamicProducerStateManagerConfig
public int producerIdExpirationMs() {
return config.getInt(PRODUCER_ID_EXPIRATION_MS_CONFIG);
}

public int producerIdExpirationCheckIntervalMs() {
return producerIdExpirationCheckIntervalMs;
}


}
11 changes: 4 additions & 7 deletions ...src/main/java/org/apache/kafka/coordinator/transaction/TransactionStateManagerConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,10 +55,10 @@ public final class TransactionStateManagerConfig {
.define(TRANSACTIONS_ABORT_TIMED_OUT_TRANSACTION_CLEANUP_INTERVAL_MS_CONFIG, INT, TRANSACTIONS_ABORT_TIMED_OUT_TRANSACTION_CLEANUP_INTERVAL_MS_DEFAULT, atLeast(1), LOW, TRANSACTIONS_ABORT_TIMED_OUT_TRANSACTIONS_INTERVAL_MS_DOC)
.define(TRANSACTIONS_REMOVE_EXPIRED_TRANSACTIONAL_ID_CLEANUP_INTERVAL_MS_CONFIG, INT, TRANSACTIONS_REMOVE_EXPIRED_TRANSACTIONAL_ID_CLEANUP_INTERVAL_MS_DEFAULT, atLeast(1), LOW, TRANSACTIONS_REMOVE_EXPIRED_TRANSACTIONS_INTERVAL_MS_DOC);

private int transactionalIdExpirationMs;
private int transactionMaxTimeoutMs;
private int transactionAbortTimedOutTransactionCleanupIntervalMs;
private int transactionRemoveExpiredTransactionalIdCleanupIntervalMs;
private final int transactionalIdExpirationMs;
private final int transactionMaxTimeoutMs;
private final int transactionAbortTimedOutTransactionCleanupIntervalMs;
private final int transactionRemoveExpiredTransactionalIdCleanupIntervalMs;

public TransactionStateManagerConfig(AbstractConfig config) {
transactionalIdExpirationMs = config.getInt(TransactionStateManagerConfig.TRANSACTIONAL_ID_EXPIRATION_MS_CONFIG);
Expand All @@ -81,7 +81,4 @@ public int transactionAbortTimedOutTransactionCleanupIntervalMs() {
public int transactionRemoveExpiredTransactionalIdCleanupIntervalMs() {
return transactionRemoveExpiredTransactionalIdCleanupIntervalMs;
}



}