You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2025/11/12 19:33:45 UTC
(iggy) branch master updated: fix(java): remove client side message checksum generation (#2342)
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new b5323d368 fix(java): remove client side message checksum generation (#2342)
b5323d368 is described below
commit b5323d368eb13a8621938c08b2bb1535c6f54ee0
Author: Maciej Modzelewski <ma...@modzelewski.dev>
AuthorDate: Wed Nov 12 20:33:40 2025 +0100
fix(java): remove client side message checksum generation (#2342)
---
.../iggy/connector/flink/sink/IggySinkWriter.java | 26 ++++++++-----------
.../main/java/org/apache/iggy/message/Message.java | 30 ++++++++--------------
2 files changed, 21 insertions(+), 35 deletions(-)
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkWriter.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkWriter.java
index e12ecb88a..5a7d85e53 100644
--- a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkWriter.java
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkWriter.java
@@ -21,14 +21,14 @@ package org.apache.iggy.connector.flink.sink;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.iggy.client.blocking.http.IggyHttpClient;
+import org.apache.iggy.connector.error.ConnectorException;
+import org.apache.iggy.connector.serialization.SerializationSchema;
import org.apache.iggy.identifier.StreamId;
import org.apache.iggy.identifier.TopicId;
import org.apache.iggy.message.Message;
import org.apache.iggy.message.MessageHeader;
import org.apache.iggy.message.MessageId;
import org.apache.iggy.message.Partitioning;
-import org.apache.iggy.connector.error.ConnectorException;
-import org.apache.iggy.connector.serialization.SerializationSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,8 +38,6 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
-import java.util.zip.CRC32;
-import java.util.zip.Checksum;
/**
* Sink writer implementation for writing records to Iggy.
@@ -73,12 +71,12 @@ public class IggySinkWriter<T> implements SinkWriter<T> {
/**
* Creates a new sink writer.
*
- * @param httpClient the HTTP Iggy client
- * @param streamId the stream identifier
- * @param topicId the topic identifier
- * @param serializer the serialization schema
- * @param batchSize maximum number of records to buffer before flushing
- * @param flushInterval maximum time to wait before flushing
+ * @param httpClient the HTTP Iggy client
+ * @param streamId the stream identifier
+ * @param topicId the topic identifier
+ * @param serializer the serialization schema
+ * @param batchSize maximum number of records to buffer before flushing
+ * @param flushInterval maximum time to wait before flushing
* @param partitioningStrategy the partitioning strategy
*/
public IggySinkWriter(
@@ -145,7 +143,7 @@ public class IggySinkWriter<T> implements SinkWriter<T> {
}
LOGGER.debug("IggySinkWriter.flush() - flushing {} messages to stream={}, topic={}",
- buffer.size(), streamId, topicId);
+ buffer.size(), streamId, topicId);
try {
// Serialize all buffered records
@@ -285,12 +283,8 @@ public class IggySinkWriter<T> implements SinkWriter<T> {
* @return a Message instance
*/
private Message createMessage(byte[] payload) {
- Checksum crc32 = new CRC32();
- crc32.update(payload, 0, payload.length);
- BigInteger checksum = BigInteger.valueOf(crc32.getValue());
-
MessageHeader header = new MessageHeader(
- checksum,
+ BigInteger.ZERO,
MessageId.serverGenerated(),
BigInteger.ZERO, // offset
BigInteger.ZERO, // timestamp
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Message.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Message.java
index 29cfd38cf..9c069445f 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Message.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Message.java
@@ -22,8 +22,6 @@ package org.apache.iggy.message;
import java.math.BigInteger;
import java.util.Map;
import java.util.Optional;
-import java.util.zip.CRC32;
-import java.util.zip.Checksum;
public record Message(
MessageHeader header,
@@ -31,22 +29,16 @@ public record Message(
Optional<Map<String, HeaderValue>> userHeaders
) {
- public static Message of(String payload) {
- final byte[] payloadBytes = payload.getBytes();
- final MessageHeader msgHeader = new MessageHeader(getChecksum(payloadBytes), MessageId.serverGenerated(),
- BigInteger.ZERO, BigInteger.ZERO, BigInteger.ZERO, 0L,
- (long)payloadBytes.length);
- return new Message(msgHeader, payloadBytes, Optional.empty());
- }
+ public static Message of(String payload) {
+ final byte[] payloadBytes = payload.getBytes();
+ final MessageHeader msgHeader = new MessageHeader(BigInteger.ZERO, MessageId.serverGenerated(),
+ BigInteger.ZERO, BigInteger.ZERO, BigInteger.ZERO, 0L,
+ (long) payloadBytes.length);
+ return new Message(msgHeader, payloadBytes, Optional.empty());
+ }
- private static BigInteger getChecksum(byte[] payload) {
- final Checksum crc32 = new CRC32();
- crc32.update(payload, 0, payload.length);
- return BigInteger.valueOf(crc32.getValue());
- }
-
- public int getSize() {
- // userHeaders is empty for now.
- return MessageHeader.SIZE + payload.length;
- }
+ public int getSize() {
+ // userHeaders is empty for now.
+ return MessageHeader.SIZE + payload.length;
+ }
}