You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mjsax (via GitHub)" <gi...@apache.org> on 2025/01/24 23:13:34 UTC
[PR] KAFKA-18644: improve generic type names for internal FK-join classes [kafka]
mjsax opened a new pull request, #18700:
URL: https://github.com/apache/kafka/pull/18700
Naming of generic types is not consistent, and hard to read. Standardizing to use `KLeft`/`VLeft` for the left/primary table, and `KRight`/`VRight` for right/foreign table, and using `VOut` as join result type.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18644: improve generic type names for internal FK-join classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #18700:
URL: https://github.com/apache/kafka/pull/18700#discussion_r1935994199
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKey.java:
##########
@@ -14,26 +14,25 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.kafka.streams.kstream.internals.foreignkeyjoin;
import java.util.Objects;
-public class CombinedKey<KF, KP> {
- private final KF foreignKey;
- private final KP primaryKey;
+public class CombinedKey<KRight, KLeft> {
Review Comment:
The order is on-purpose, because that is the actually order of both arguments of `CombinedKey`, ie, especially how we serialize it (but also constructor argument order; and the constructor argument order is also the way it is, to express the serialization order), what is important as we apply range-scans on RocksDB using the combined-key. I was also first thinking about maybe swapping it, but in the end, I did keep as is, because it might be better to align the order?
About the names. I did the change on-purpose, because the variable names, already have good names, it it IMHO make it's very clear?
```
private final KRight foreignKey;
private final KLeft primaryKey;
```
plus:
```
CombinedKey(final KRight foreignKey, final KLeft primaryKey)
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18644: improve generic type names for internal FK-join classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #18700:
URL: https://github.com/apache/kafka/pull/18700
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18644: improve generic type names for internal FK-join classes [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18700:
URL: https://github.com/apache/kafka/pull/18700#discussion_r1935249553
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKey.java:
##########
@@ -14,26 +14,25 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.kafka.streams.kstream.internals.foreignkeyjoin;
import java.util.Objects;
-public class CombinedKey<KF, KP> {
- private final KF foreignKey;
- private final KP primaryKey;
+public class CombinedKey<KRight, KLeft> {
Review Comment:
It's a bit weird that right is left and left is right. But not sure if I have a solution as we cannot swap them. How about standardizing on `KForeign` and `KPrimary`? That seemed rather clear to me, but I may be lacking context.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18644: improve generic type names for internal FK-join classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #18700:
URL: https://github.com/apache/kafka/pull/18700#discussion_r1929357204
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchema.java:
##########
@@ -50,17 +50,17 @@ public CombinedKeySchema(final Supplier<String> foreignKeySerdeTopicSupplier,
foreignKeySerializer = foreignKeySerde == null ? null : foreignKeySerde.serializer();
}
- @SuppressWarnings("unchecked")
+ @SuppressWarnings({"unchecked", "resource"})
Review Comment:
Suppression some IntellIj warnings about "potential resource leaks" for not using try-with-resource for the "Closable" Serdes we get.
Similar elsewhere.
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionJoinProcessorSupplier.java:
##########
@@ -31,53 +30,44 @@
import java.util.Objects;
/**
- * Receives {@code SubscriptionWrapper<K>} events and processes them according to their Instruction.
+ * Receives {@code SubscriptionWrapper<KLeft>} events and processes them according to their Instruction.
* Depending on the results, {@code SubscriptionResponseWrapper}s are created, which will be propagated to
* the {@code ResponseJoinProcessorSupplier} instance.
*
- * @param <K> Type of primary keys
- * @param <KO> Type of foreign key
- * @param <VO> Type of foreign value
+ * @param <KLeft> Type of primary keys
+ * @param <KRight> Type of foreign key
+ * @param <VRight> Type of foreign value
*/
-public class SubscriptionJoinProcessorSupplier<K, KO, VO>
- implements ProcessorSupplier<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>, K, SubscriptionResponseWrapper<VO>> {
+public class SubscriptionJoinProcessorSupplier<KLeft, KRight, VRight>
+ implements ProcessorSupplier<CombinedKey<KRight, KLeft>, Change<ValueAndTimestamp<SubscriptionWrapper<KLeft>>>, KLeft, SubscriptionResponseWrapper<VRight>> {
- private final KTableValueGetterSupplier<KO, VO> foreignValueGetterSupplier;
+ private final KTableValueGetterSupplier<KRight, VRight> foreignValueGetterSupplier;
- public SubscriptionJoinProcessorSupplier(final KTableValueGetterSupplier<KO, VO> foreignValueGetterSupplier) {
+ public SubscriptionJoinProcessorSupplier(final KTableValueGetterSupplier<KRight, VRight> foreignValueGetterSupplier) {
this.foreignValueGetterSupplier = foreignValueGetterSupplier;
}
@Override
- public Processor<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>, K, SubscriptionResponseWrapper<VO>> get() {
-
- return new ContextualProcessor<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>, K, SubscriptionResponseWrapper<VO>>() {
-
- private KTableValueGetter<KO, VO> foreignValues;
+ public Processor<CombinedKey<KRight, KLeft>, Change<ValueAndTimestamp<SubscriptionWrapper<KLeft>>>, KLeft, SubscriptionResponseWrapper<VRight>> get() {
+ return new ContextualProcessor<>() {
+ private KTableValueGetter<KRight, VRight> foreignValues;
@Override
- public void init(final ProcessorContext<K, SubscriptionResponseWrapper<VO>> context) {
+ public void init(final ProcessorContext<KLeft, SubscriptionResponseWrapper<VRight>> context) {
super.init(context);
foreignValues = foreignValueGetterSupplier.get();
foreignValues.init(context);
}
@Override
- public void process(final Record<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>> record) {
+ public void process(final Record<CombinedKey<KRight, KLeft>, Change<ValueAndTimestamp<SubscriptionWrapper<KLeft>>>> record) {
Objects.requireNonNull(record.key(), "This processor should never see a null key.");
Objects.requireNonNull(record.value(), "This processor should never see a null value.");
- final ValueAndTimestamp<SubscriptionWrapper<K>> valueAndTimestamp = record.value().newValue;
+ final ValueAndTimestamp<SubscriptionWrapper<KLeft>> valueAndTimestamp = record.value().newValue;
Objects.requireNonNull(valueAndTimestamp, "This processor should never see a null newValue.");
- final SubscriptionWrapper<K> value = valueAndTimestamp.value();
-
- if (value.version() > SubscriptionWrapper.CURRENT_VERSION) {
- //Guard against modifications to SubscriptionWrapper. Need to ensure that there is compatibility
- //with previous versions to enable rolling upgrades. Must develop a strategy for upgrading
- //from older SubscriptionWrapper versions to newer versions.
- throw new UnsupportedVersionException("SubscriptionWrapper is of an incompatible version.");
- }
+ final SubscriptionWrapper<KLeft> value = subscriptionWrapper(valueAndTimestamp);
Review Comment:
Small refactoring proposed by IntelliJ -- guess it make sense to extract into a helper method -- easier to read overall, as semantics are a little clearer.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18644: improve generic type names for internal FK-join classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #18700:
URL: https://github.com/apache/kafka/pull/18700#issuecomment-2614156367
Java 17:
```
FAILED ❌ CooperativeConsumerCoordinatorTest > testOutdatedCoordinatorAssignment()
FAILED ❌ SharePartitionManagerTest > testReplicaManagerFetchShouldProceed()
FAILED ❌ SharePartitionManagerTest > testDelayedInitializationShouldCompleteFetchRequest()
FAILED ❌ RemoteIndexCacheTest > testIndexFileAlreadyExistOnDiskButNotInCache()
Found 3 flaky test failures:
FLAKY ⚠️ KafkaShareConsumerTest > testVerifyFetchAndAcknowledgeSync()
FLAKY ⚠️ AbstractCoordinatorTest > testWakeupAfterSyncGroupReceivedExternalCompletion()
FLAKY ⚠️ StickyAssignorTest > testLargeAssignmentAndGroupWithNonEqualSubscription(boolean).hasConsumerRack = false
```
Java 23:
```
Found 3 test failures:
FAILED ❌ SharePartitionManagerTest > testReplicaManagerFetchShouldProceed()
FAILED ❌ SharePartitionManagerTest > testDelayedInitializationShouldCompleteFetchRequest()
FAILED ❌ MetricsDuringTopicCreationDeletionTest > "testMetricsDuringTopicCreateDelete(String).quorum=kraft"
Found 1 flaky test failures:
FLAKY ⚠️ StickyAssignorTest > testLargeAssignmentAndGroupWithNonEqualSubscription(boolean).hasConsumerRack = false
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18644: improve generic type names for internal FK-join classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #18700:
URL: https://github.com/apache/kafka/pull/18700#discussion_r1935994199
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKey.java:
##########
@@ -14,26 +14,25 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.kafka.streams.kstream.internals.foreignkeyjoin;
import java.util.Objects;
-public class CombinedKey<KF, KP> {
- private final KF foreignKey;
- private final KP primaryKey;
+public class CombinedKey<KRight, KLeft> {
Review Comment:
The order is on-purpose, because that is the actually order of both arguments of `CombinedKey`, ie, especially how we serialize it. I was also first thinking about maybe swapping it, but in the end, I did keep as, as it might be better to align the order to the serialization order?
Let me think about the names a little bit more, so it can sink it :)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18644: improve generic type names for internal FK-join classes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #18700:
URL: https://github.com/apache/kafka/pull/18700#issuecomment-2628561126
Java 17
```
Found 1 test failures:
FAILED ❌ CooperativeConsumerCoordinatorTest > testOutdatedCoordinatorAssignment()
Found 2 flaky test failures:
FLAKY ⚠️ AbstractCoordinatorTest > testWakeupAfterSyncGroupReceivedExternalCompletion()
FLAKY ⚠️ TransactionsWithTieredStoreTest > testFencingOnTransactionExpiration(String, String).quorum=kraft.groupProtocol=consumer
```
Java 23
```
Found 3 test failures:
FAILED ❌ ConfigCommandIntegrationTest > testUpdateInvalidBrokerConfigs [2] Type=Raft-Combined, MetadataVersion=4.0-IV3,BrokerSecurityProtocol=PLAINTEXT,BrokerListenerName=ListenerName(EXTERNAL),ControllerSecurityProtocol=PLAINTEXT,ControllerListenerName=ListenerName(CONTROLLER)
FAILED ❌ TransactionsTest > testFencingOnTransactionExpiration(String, String).quorum=kraft.groupProtocol=classic
FAILED ❌ ClientQuotasRequestTest > testClientQuotasWithDefaultName [1] Type=Raft-Isolated, MetadataVersion=4.0-IV3,BrokerSecurityProtocol=PLAINTEXT,BrokerListenerName=ListenerName(EXTERNAL),ControllerSecurityProtocol=PLAINTEXT,ControllerListenerName=ListenerName(CONTROLLER)
Found 1 flaky test failures:
FLAKY ⚠️ TransactionsWithTieredStoreTest > testFencingOnTransactionExpiration(String, String).quorum=kraft.groupProtocol=classic
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18644: improve generic type names for internal FK-join classes [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18700:
URL: https://github.com/apache/kafka/pull/18700#discussion_r1939033861
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKey.java:
##########
@@ -14,26 +14,25 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.kafka.streams.kstream.internals.foreignkeyjoin;
import java.util.Objects;
-public class CombinedKey<KF, KP> {
- private final KF foreignKey;
- private final KP primaryKey;
+public class CombinedKey<KRight, KLeft> {
Review Comment:
I agree on keeping the order as it is serialized, that's why I said we shouldn't swap them. I think I have a slight preference for using KForeign and KPrimary, but I'm also fine with keeping KRight/KLeft. It's true that you can use the variable names to discern that.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org