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 2024/09/16 02:37:58 UTC
[PR] MINOR: Fix SubscriptionResponseWrapperSerializer [kafka]
mjsax opened a new pull request, #17205:
URL: https://github.com/apache/kafka/pull/17205
(no comment)
--
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] MINOR: Fix SubscriptionResponseWrapperSerializer [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17205:
URL: https://github.com/apache/kafka/pull/17205#discussion_r1772476886
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapperSerdeTest.java:
##########
@@ -73,68 +69,95 @@ public T deserialize(final String topic, final byte[] data) {
}
@Test
- @SuppressWarnings("unchecked")
public void ShouldSerdeWithNonNullsTest() {
final long[] hashedValue = Murmur3.hash128(new byte[] {(byte) 0x01, (byte) 0x9A, (byte) 0xFF, (byte) 0x00});
final String foreignValue = "foreignValue";
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, foreignValue, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertEquals(foreignValue, result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertEquals(foreignValue, result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
- @SuppressWarnings("unchecked")
public void shouldSerdeWithNullForeignValueTest() {
final long[] hashedValue = Murmur3.hash128(new byte[] {(byte) 0x01, (byte) 0x9A, (byte) 0xFF, (byte) 0x00});
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, null, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertNull(result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertNull(result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
- @SuppressWarnings("unchecked")
public void shouldSerdeWithNullHashTest() {
final long[] hashedValue = null;
final String foreignValue = "foreignValue";
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, foreignValue, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertEquals(foreignValue, result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertEquals(foreignValue, result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
- @SuppressWarnings("unchecked")
public void shouldSerdeWithNullsTest() {
final long[] hashedValue = null;
final String foreignValue = null;
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, foreignValue, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertEquals(foreignValue, result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertEquals(foreignValue, result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
public void shouldThrowExceptionWithBadVersionTest() {
final long[] hashedValue = null;
- assertThrows(UnsupportedVersionException.class,
- () -> new SubscriptionResponseWrapper<>(hashedValue, "foreignValue", (byte) 0xFF, 1));
+ assertThrows(
+ UnsupportedVersionException.class,
+ () -> new SubscriptionResponseWrapper<>(hashedValue, "foreignValue", (byte) -1, 1)
Review Comment:
Does not make much difference. Will just merge as-is to avoid another Jenkins run.
--
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] MINOR: Fix SubscriptionResponseWrapperSerializer [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #17205:
URL: https://github.com/apache/kafka/pull/17205#issuecomment-2369989314
Thanks for the review @chia7712. Merged to `trunk`.
--
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] MINOR: Fix SubscriptionResponseWrapperSerializer [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17205:
URL: https://github.com/apache/kafka/pull/17205#discussion_r1760466560
##########
streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java:
##########
@@ -46,7 +46,6 @@ default ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], b
* @param record The record that failed to produce
* @param exception The exception that occurred during production
*/
- @SuppressWarnings("deprecation")
Review Comment:
Side cleanup.
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapperSerdeTest.java:
##########
@@ -73,68 +69,95 @@ public T deserialize(final String topic, final byte[] data) {
}
@Test
- @SuppressWarnings("unchecked")
public void ShouldSerdeWithNonNullsTest() {
final long[] hashedValue = Murmur3.hash128(new byte[] {(byte) 0x01, (byte) 0x9A, (byte) 0xFF, (byte) 0x00});
final String foreignValue = "foreignValue";
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, foreignValue, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertEquals(foreignValue, result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertEquals(foreignValue, result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
- @SuppressWarnings("unchecked")
public void shouldSerdeWithNullForeignValueTest() {
final long[] hashedValue = Murmur3.hash128(new byte[] {(byte) 0x01, (byte) 0x9A, (byte) 0xFF, (byte) 0x00});
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, null, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertNull(result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertNull(result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
- @SuppressWarnings("unchecked")
public void shouldSerdeWithNullHashTest() {
final long[] hashedValue = null;
final String foreignValue = "foreignValue";
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, foreignValue, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertEquals(foreignValue, result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertEquals(foreignValue, result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
- @SuppressWarnings("unchecked")
public void shouldSerdeWithNullsTest() {
final long[] hashedValue = null;
final String foreignValue = null;
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, foreignValue, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertEquals(foreignValue, result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertEquals(foreignValue, result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
public void shouldThrowExceptionWithBadVersionTest() {
final long[] hashedValue = null;
- assertThrows(UnsupportedVersionException.class,
- () -> new SubscriptionResponseWrapper<>(hashedValue, "foreignValue", (byte) 0xFF, 1));
+ assertThrows(
+ UnsupportedVersionException.class,
+ () -> new SubscriptionResponseWrapper<>(hashedValue, "foreignValue", (byte) -1, 1)
Review Comment:
Minor change for this test, too, passing in `-1` now to make it more logical.
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapperSerde.java:
##########
@@ -66,9 +66,8 @@ public void setIfUnset(final SerdeGetter getter) {
public byte[] serialize(final String topic, final SubscriptionResponseWrapper<V> data) {
//{1-bit-isHashNull}{7-bits-version}{Optional-16-byte-Hash}{n-bytes serialized data}
- //7-bit (0x7F) maximum for data version.
- if (Byte.compare((byte) 0x7F, data.version()) < 0) {
Review Comment:
This check is always `false`. Byte has a range from `-128...127`; thus, this comparison is incorrect.
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapperSerdeTest.java:
##########
@@ -73,68 +69,95 @@ public T deserialize(final String topic, final byte[] data) {
}
@Test
- @SuppressWarnings("unchecked")
Review Comment:
Side cleanup... (same other tests below)
--
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] MINOR: Fix SubscriptionResponseWrapperSerializer [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #17205:
URL: https://github.com/apache/kafka/pull/17205
--
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] MINOR: Fix SubscriptionResponseWrapperSerializer [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17205:
URL: https://github.com/apache/kafka/pull/17205#discussion_r1769380792
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapperSerdeTest.java:
##########
@@ -73,68 +69,95 @@ public T deserialize(final String topic, final byte[] data) {
}
@Test
- @SuppressWarnings("unchecked")
public void ShouldSerdeWithNonNullsTest() {
final long[] hashedValue = Murmur3.hash128(new byte[] {(byte) 0x01, (byte) 0x9A, (byte) 0xFF, (byte) 0x00});
final String foreignValue = "foreignValue";
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, foreignValue, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertEquals(foreignValue, result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertEquals(foreignValue, result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
- @SuppressWarnings("unchecked")
public void shouldSerdeWithNullForeignValueTest() {
final long[] hashedValue = Murmur3.hash128(new byte[] {(byte) 0x01, (byte) 0x9A, (byte) 0xFF, (byte) 0x00});
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, null, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertNull(result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertNull(result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
- @SuppressWarnings("unchecked")
public void shouldSerdeWithNullHashTest() {
final long[] hashedValue = null;
final String foreignValue = "foreignValue";
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, foreignValue, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertEquals(foreignValue, result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertEquals(foreignValue, result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
- @SuppressWarnings("unchecked")
public void shouldSerdeWithNullsTest() {
final long[] hashedValue = null;
final String foreignValue = null;
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, foreignValue, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertEquals(foreignValue, result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertEquals(foreignValue, result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
public void shouldThrowExceptionWithBadVersionTest() {
final long[] hashedValue = null;
- assertThrows(UnsupportedVersionException.class,
- () -> new SubscriptionResponseWrapper<>(hashedValue, "foreignValue", (byte) 0xFF, 1));
+ assertThrows(
+ UnsupportedVersionException.class,
+ () -> new SubscriptionResponseWrapper<>(hashedValue, "foreignValue", (byte) -1, 1)
Review Comment:
Maybe `hashedValue` can be replaced by `null`
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapperSerdeTest.java:
##########
@@ -73,68 +69,95 @@ public T deserialize(final String topic, final byte[] data) {
}
@Test
- @SuppressWarnings("unchecked")
public void ShouldSerdeWithNonNullsTest() {
final long[] hashedValue = Murmur3.hash128(new byte[] {(byte) 0x01, (byte) 0x9A, (byte) 0xFF, (byte) 0x00});
final String foreignValue = "foreignValue";
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, foreignValue, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertEquals(foreignValue, result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertEquals(foreignValue, result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
- @SuppressWarnings("unchecked")
public void shouldSerdeWithNullForeignValueTest() {
final long[] hashedValue = Murmur3.hash128(new byte[] {(byte) 0x01, (byte) 0x9A, (byte) 0xFF, (byte) 0x00});
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, null, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertNull(result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertNull(result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
- @SuppressWarnings("unchecked")
public void shouldSerdeWithNullHashTest() {
final long[] hashedValue = null;
final String foreignValue = "foreignValue";
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, foreignValue, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertEquals(foreignValue, result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertEquals(foreignValue, result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
- @SuppressWarnings("unchecked")
public void shouldSerdeWithNullsTest() {
final long[] hashedValue = null;
final String foreignValue = null;
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, foreignValue, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertEquals(foreignValue, result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertEquals(foreignValue, result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
public void shouldThrowExceptionWithBadVersionTest() {
final long[] hashedValue = null;
- assertThrows(UnsupportedVersionException.class,
- () -> new SubscriptionResponseWrapper<>(hashedValue, "foreignValue", (byte) 0xFF, 1));
+ assertThrows(
+ UnsupportedVersionException.class,
+ () -> new SubscriptionResponseWrapper<>(hashedValue, "foreignValue", (byte) -1, 1)
+ );
+ }
+
+ @Test
+ public void shouldThrowExceptionOnSerializeWhenDataVersionUnknown() {
+ final SubscriptionResponseWrapper<String> srw = new InvalidSubscriptionResponseWrapper(null, null, 1);
Review Comment:
How about using anonymous class?
```java
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<String>(null, null, 1) {
@Override
public byte version() {
return -1;
}
};
```
--
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] MINOR: Fix SubscriptionResponseWrapperSerializer [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17205:
URL: https://github.com/apache/kafka/pull/17205#discussion_r1772476702
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapperSerdeTest.java:
##########
@@ -73,68 +69,95 @@ public T deserialize(final String topic, final byte[] data) {
}
@Test
- @SuppressWarnings("unchecked")
public void ShouldSerdeWithNonNullsTest() {
final long[] hashedValue = Murmur3.hash128(new byte[] {(byte) 0x01, (byte) 0x9A, (byte) 0xFF, (byte) 0x00});
final String foreignValue = "foreignValue";
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, foreignValue, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertEquals(foreignValue, result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertEquals(foreignValue, result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
- @SuppressWarnings("unchecked")
public void shouldSerdeWithNullForeignValueTest() {
final long[] hashedValue = Murmur3.hash128(new byte[] {(byte) 0x01, (byte) 0x9A, (byte) 0xFF, (byte) 0x00});
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, null, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertNull(result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertNull(result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
- @SuppressWarnings("unchecked")
public void shouldSerdeWithNullHashTest() {
final long[] hashedValue = null;
final String foreignValue = "foreignValue";
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, foreignValue, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertEquals(foreignValue, result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertEquals(foreignValue, result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
- @SuppressWarnings("unchecked")
public void shouldSerdeWithNullsTest() {
final long[] hashedValue = null;
final String foreignValue = null;
final SubscriptionResponseWrapper<String> srw = new SubscriptionResponseWrapper<>(hashedValue, foreignValue, 1);
- final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde(new NonNullableSerde(Serdes.String()));
- final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
- final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
+ try (final SubscriptionResponseWrapperSerde<String> srwSerde = new SubscriptionResponseWrapperSerde<>(new NonNullableSerde<>(Serdes.String()))) {
+ final byte[] serResponse = srwSerde.serializer().serialize(null, srw);
+ final SubscriptionResponseWrapper<String> result = srwSerde.deserializer().deserialize(null, serResponse);
- assertArrayEquals(hashedValue, result.originalValueHash());
- assertEquals(foreignValue, result.foreignValue());
- assertNull(result.primaryPartition());
+ assertArrayEquals(hashedValue, result.originalValueHash());
+ assertEquals(foreignValue, result.foreignValue());
+ assertNull(result.primaryPartition());
+ }
}
@Test
public void shouldThrowExceptionWithBadVersionTest() {
final long[] hashedValue = null;
- assertThrows(UnsupportedVersionException.class,
- () -> new SubscriptionResponseWrapper<>(hashedValue, "foreignValue", (byte) 0xFF, 1));
+ assertThrows(
+ UnsupportedVersionException.class,
+ () -> new SubscriptionResponseWrapper<>(hashedValue, "foreignValue", (byte) -1, 1)
+ );
+ }
+
+ @Test
+ public void shouldThrowExceptionOnSerializeWhenDataVersionUnknown() {
+ final SubscriptionResponseWrapper<String> srw = new InvalidSubscriptionResponseWrapper(null, null, 1);
Review Comment:
Using a named class explains the semantics of the test better. Think it better as-is.
--
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