root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-10618: Rename UUID to Uuid and make it more efficient (#9566) 

As decided in KIP-516, the UUID class should be named Uuid. Change all instances of
org.apache.kafka.common.UUID to org.apache.kafka.common.Uuid.

Also modify Uuid so that it stores two `long` fields instead of wrapping java.util.UUID
to reduce memory usage.

Reviewers: Ismael Juma <ismael@juma.me.uk>
This commit is contained in:
Justine Olshan 2020-11-18 03:58:20 -05:00 committed by GitHub
parent 8c163e027e
commit 28c57b273a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 102 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
clients/src/main/java/org/apache/kafka/common/UUID.java → clients/src/main/java/org/apache/kafka/common/Uuid.java
View File

@ -22,68 +22,66 @@ import java.util.Base64;
/**
* This class defines an immutable universally unique identifier (UUID). It represents a 128-bit value.
* More specifically, the random UUIDs generated by this class are variant 2 (Leach-Salz) version 4 UUIDs.
* This is the same type of UUID as the ones generated by java.util.UUID except that the toString() method prints
* This is the same type of UUID as the ones generated by java.util.UUID. The toString() method prints
* using the base64 string encoding. Likewise, the fromString method expects a base64 string encoding.
*/
public class UUID {
public class Uuid {
private static final java.util.UUID SENTINEL_ID_INTERNAL = new java.util.UUID(0L, 1L);
/**
* A UUID that represents a null or empty UUID. Will never be returned by the randomUUID method
* A UUID that represents a null or empty UUID. Will never be returned by the randomUuid method.
*/
public static final UUID ZERO_UUID = new UUID(new java.util.UUID(0L, 0L));
public static final Uuid ZERO_UUID = new Uuid(0L, 0L);
private static final java.util.UUID ZERO_ID_INTERNAL = new java.util.UUID(0L, 0L);
private final java.util.UUID uuid;
private final long mostSignificantBits;
private final long leastSignificantBits;
/**
* Constructs a 128-bit type 4 UUID where the first long represents the the most significant 64 bits
* and the second long represents the least significant 64 bits.
*/
public UUID(long mostSigBits, long leastSigBits) {
this.uuid = new java.util.UUID(mostSigBits, leastSigBits);
}
private UUID(java.util.UUID uuid) {
this.uuid = uuid;
public Uuid(long mostSigBits, long leastSigBits) {
this.mostSignificantBits = mostSigBits;
this.leastSignificantBits = leastSigBits;
}
/**
* Static factory to retrieve a type 4 (pseudo randomly generated) UUID.
*/
public static UUID randomUUID() {
public static Uuid randomUuid() {
java.util.UUID uuid = java.util.UUID.randomUUID();
while (uuid.equals(SENTINEL_ID_INTERNAL) || uuid.equals(ZERO_ID_INTERNAL)) {
uuid = java.util.UUID.randomUUID();
}
return new UUID(uuid);
return new Uuid(uuid.getMostSignificantBits(), uuid.getLeastSignificantBits());
}
/**
* Returns the most significant bits of the UUID's 128 value.
*/
public long getMostSignificantBits() {
return uuid.getMostSignificantBits();
return this.mostSignificantBits;
}
/**
* Returns the least significant bits of the UUID's 128 value.
*/
public long getLeastSignificantBits() {
return uuid.getLeastSignificantBits();
return this.leastSignificantBits;
}
/**
* Returns true iff obj is another UUID represented by the same two long values.
* Returns true iff obj is another Uuid represented by the same two long values.
*/
@Override
public boolean equals(Object obj) {
if ((null == obj) || (obj.getClass() != this.getClass()))
return false;
UUID id = (UUID) obj;
return this.getMostSignificantBits() == id.getMostSignificantBits() &&
this.getLeastSignificantBits() == id.getLeastSignificantBits();
Uuid id = (Uuid) obj;
return this.mostSignificantBits == id.mostSignificantBits &&
this.leastSignificantBits == id.leastSignificantBits;
}
/**
@ -91,7 +89,8 @@ public class UUID {
*/
@Override
public int hashCode() {
return uuid.hashCode();
long xor = mostSignificantBits ^ leastSignificantBits;
return (int) (xor >> 32) ^ (int) xor;
}
/**
@ -99,22 +98,22 @@ public class UUID {
*/
@Override
public String toString() {
return Base64.getUrlEncoder().withoutPadding().encodeToString(getBytesFromUuid(uuid));
return Base64.getUrlEncoder().withoutPadding().encodeToString(getBytesFromUuid());
}
/**
* Creates a UUID based on a base64 string encoding used in the toString() method.
*/
public static UUID fromString(String str) {
public static Uuid fromString(String str) {
ByteBuffer uuidBytes = ByteBuffer.wrap(Base64.getUrlDecoder().decode(str));
return new UUID(uuidBytes.getLong(), uuidBytes.getLong());
return new Uuid(uuidBytes.getLong(), uuidBytes.getLong());
}
private byte[] getBytesFromUuid(java.util.UUID uuid) {
private byte[] getBytesFromUuid() {
// Extract bytes for uuid which is 128 bits (or 16 bytes) long.
ByteBuffer uuidBytes = ByteBuffer.wrap(new byte[16]);
uuidBytes.putLong(uuid.getMostSignificantBits());
uuidBytes.putLong(uuid.getLeastSignificantBits());
uuidBytes.putLong(this.mostSignificantBits);
uuidBytes.putLong(this.leastSignificantBits);
return uuidBytes.array();
}
}

6
clients/src/main/java/org/apache/kafka/common/protocol/Readable.java
View File

@ -17,7 +17,7 @@
package org.apache.kafka.common.protocol;
import org.apache.kafka.common.UUID;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.protocol.types.RawTaggedField;
import org.apache.kafka.common.record.MemoryRecords;
@ -67,7 +67,7 @@ public interface Readable {
/**
* Read a UUID with the most significant digits first.
*/
default UUID readUUID() {
return new UUID(readLong(), readLong());
default Uuid readUuid() {
return new Uuid(readLong(), readLong());
}
}

4
clients/src/main/java/org/apache/kafka/common/protocol/Writable.java
View File

@ -17,7 +17,7 @@
package org.apache.kafka.common.protocol;
import org.apache.kafka.common.UUID;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.record.BaseRecords;
import org.apache.kafka.common.record.MemoryRecords;
@ -44,7 +44,7 @@ public interface Writable {
}
}
default void writeUUID(UUID uuid) {
default void writeUuid(Uuid uuid) {
writeLong(uuid.getMostSignificantBits());
writeLong(uuid.getLeastSignificantBits());
}

20
clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java
View File

@ -16,7 +16,7 @@
*/
package org.apache.kafka.common.protocol.types;
import org.apache.kafka.common.UUID;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.record.BaseRecords;
import java.nio.ByteBuffer;
@ -89,8 +89,8 @@ public class Struct {
return getLong(field.name);
}
public UUID get(Field.UUID field) {
return getUUID(field.name);
public Uuid get(Field.UUID field) {
return getUuid(field.name);
}
public Short get(Field.Int16 field) {
@ -127,9 +127,9 @@ public class Struct {
return alternative;
}
public UUID getOrElse(Field.UUID field, UUID alternative) {
public Uuid getOrElse(Field.UUID field, Uuid alternative) {
if (hasField(field.name))
return getUUID(field.name);
return getUuid(field.name);
return alternative;
}
@ -266,12 +266,12 @@ public class Struct {
return (Long) get(name);
}
public UUID getUUID(BoundField field) {
return (UUID) get(field);
public Uuid getUuid(BoundField field) {
return (Uuid) get(field);
}
public UUID getUUID(String name) {
return (UUID) get(name);
public Uuid getUuid(String name) {
return (Uuid) get(name);
}
public Double getDouble(BoundField field) {
@ -379,7 +379,7 @@ public class Struct {
return set(def.name, value);
}
public Struct set(Field.UUID def, UUID value) {
public Struct set(Field.UUID def, Uuid value) {
return set(def.name, value);
}

16
clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java
View File

@ -16,7 +16,7 @@
*/
package org.apache.kafka.common.protocol.types;
import org.apache.kafka.common.UUID;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.record.BaseRecords;
import org.apache.kafka.common.record.MemoryRecords;
import org.apache.kafka.common.utils.ByteUtils;
@ -331,14 +331,14 @@ public abstract class Type {
public static final DocumentedType UUID = new DocumentedType() {
@Override
public void write(ByteBuffer buffer, Object o) {
final UUID uuid = (UUID) o;
final Uuid uuid = (Uuid) o;
buffer.putLong(uuid.getMostSignificantBits());
buffer.putLong(uuid.getLeastSignificantBits());
}
@Override
public Object read(ByteBuffer buffer) {
return new UUID(buffer.getLong(), buffer.getLong());
return new Uuid(buffer.getLong(), buffer.getLong());
}
@Override
@ -352,16 +352,16 @@ public abstract class Type {
}
@Override
public UUID validate(Object item) {
if (item instanceof UUID)
return (UUID) item;
public Uuid validate(Object item) {
if (item instanceof Uuid)
return (Uuid) item;
else
throw new SchemaException(item + " is not a UUID.");
throw new SchemaException(item + " is not a Uuid.");
}
@Override
public String documentation() {
return "Represents a type 4 immutable universally unique identifier (UUID). " +
return "Represents a type 4 immutable universally unique identifier (Uuid). " +
"The values are encoded using sixteen bytes in network byte order (big-endian).";
}
};

53
clients/src/test/java/org/apache/kafka/common/UUIDTest.java → clients/src/test/java/org/apache/kafka/common/UuidTest.java
View File

@ -21,50 +21,61 @@ import org.junit.Test;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
public class UUIDTest {
public class UuidTest {
@Test
public void testSignificantBits() {
UUID id = new UUID(34L, 98L);
Uuid id = new Uuid(34L, 98L);
assertEquals(id.getMostSignificantBits(), 34L);
assertEquals(id.getLeastSignificantBits(), 98L);
}
@Test
public void testUUIDEquality() {
UUID id1 = new UUID(12L, 13L);
UUID id2 = new UUID(12L, 13L);
UUID id3 = new UUID(24L, 38L);
public void testUuidEquality() {
Uuid id1 = new Uuid(12L, 13L);
Uuid id2 = new Uuid(12L, 13L);
Uuid id3 = new Uuid(24L, 38L);
assertEquals(UUID.ZERO_UUID, UUID.ZERO_UUID);
assertEquals(Uuid.ZERO_UUID, Uuid.ZERO_UUID);
assertEquals(id1, id2);
assertNotEquals(id1, id3);
assertEquals(UUID.ZERO_UUID.hashCode(), UUID.ZERO_UUID.hashCode());
assertEquals(Uuid.ZERO_UUID.hashCode(), Uuid.ZERO_UUID.hashCode());
assertEquals(id1.hashCode(), id2.hashCode());
assertNotEquals(id1.hashCode(), id3.hashCode());
}
@Test
public void testStringConversion() {
UUID id = UUID.randomUUID();
String idString = id.toString();
public void testHashCode() {
Uuid id1 = new Uuid(16L, 7L);
Uuid id2 = new Uuid(1043L, 20075L);
Uuid id3 = new Uuid(104312423523523L, 200732425676585L);
assertEquals(UUID.fromString(idString), id);
String zeroIdString = UUID.ZERO_UUID.toString();
assertEquals(UUID.fromString(zeroIdString), UUID.ZERO_UUID);
assertEquals(23, id1.hashCode());
assertEquals(19064, id2.hashCode());
assertEquals(-2011255899, id3.hashCode());
}
@Test
public void testRandomUUID() {
UUID randomID = UUID.randomUUID();
// reservedSentinel is based on the value of SENTINEL_ID_INTERNAL in UUID.
UUID reservedSentinel = new UUID(0L, 1L);
public void testStringConversion() {
Uuid id = Uuid.randomUuid();
String idString = id.toString();
assertNotEquals(randomID, UUID.ZERO_UUID);
assertEquals(Uuid.fromString(idString), id);
String zeroIdString = Uuid.ZERO_UUID.toString();
assertEquals(Uuid.fromString(zeroIdString), Uuid.ZERO_UUID);
}
@Test
public void testRandomUuid() {
Uuid randomID = Uuid.randomUuid();
// reservedSentinel is based on the value of SENTINEL_ID_INTERNAL in Uuid.
Uuid reservedSentinel = new Uuid(0L, 1L);
assertNotEquals(randomID, Uuid.ZERO_UUID);
assertNotEquals(randomID, reservedSentinel);
}
}

4
clients/src/test/java/org/apache/kafka/common/message/MessageTest.java
View File

@ -20,7 +20,7 @@ package org.apache.kafka.common.message;
import com.fasterxml.jackson.databind.JsonNode;
import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.UUID;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.message.AddPartitionsToTxnRequestData.AddPartitionsToTxnTopic;
import org.apache.kafka.common.message.AddPartitionsToTxnRequestData.AddPartitionsToTxnTopicCollection;
@ -744,7 +744,7 @@ public final class MessageTest {
));
message.setMyTaggedStruct(new SimpleExampleMessageData.TaggedStruct().setStructId("abc"));
message.setProcessId(UUID.randomUUID());
message.setProcessId(Uuid.randomUuid());
message.setMyNullableString("notNull");
message.setMyInt16((short) 3);
message.setMyString("test string");

24
clients/src/test/java/org/apache/kafka/common/message/SimpleExampleMessageTest.java
View File

@ -17,7 +17,7 @@
package org.apache.kafka.common.message;
import com.fasterxml.jackson.databind.JsonNode;
import org.apache.kafka.common.UUID;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.protocol.ByteBufferAccessor;
import org.apache.kafka.common.protocol.ObjectSerializationCache;
@ -41,7 +41,7 @@ public class SimpleExampleMessageTest {
@Test
public void shouldStoreField() {
final UUID uuid = UUID.randomUUID();
final Uuid uuid = Uuid.randomUuid();
final ByteBuffer buf = ByteBuffer.wrap(new byte[] {1, 2, 3});
final SimpleExampleMessageData out = new SimpleExampleMessageData();
@ -61,7 +61,7 @@ public class SimpleExampleMessageTest {
public void shouldThrowIfCannotWriteNonIgnorableField() {
// processId is not supported in v0 and is not marked as ignorable
final SimpleExampleMessageData out = new SimpleExampleMessageData().setProcessId(UUID.randomUUID());
final SimpleExampleMessageData out = new SimpleExampleMessageData().setProcessId(Uuid.randomUuid());
assertThrows(UnsupportedVersionException.class, () ->
out.write(new ByteBufferAccessor(ByteBuffer.allocate(64)), new ObjectSerializationCache(), (short) 0));
assertThrows(UnsupportedVersionException.class, () -> out.toStruct((short) 0));
@ -70,14 +70,14 @@ public class SimpleExampleMessageTest {
@Test
public void shouldDefaultField() {
final SimpleExampleMessageData out = new SimpleExampleMessageData();
assertEquals(UUID.fromString("AAAAAAAAAAAAAAAAAAAAAA"), out.processId());
assertEquals(Uuid.fromString("AAAAAAAAAAAAAAAAAAAAAA"), out.processId());
assertEquals(ByteUtils.EMPTY_BUF, out.zeroCopyByteBuffer());
assertEquals(ByteUtils.EMPTY_BUF, out.nullableZeroCopyByteBuffer());
}
@Test
public void shouldRoundTripFieldThroughStruct() {
final UUID uuid = UUID.randomUUID();
final Uuid uuid = Uuid.randomUuid();
final ByteBuffer buf = ByteBuffer.wrap(new byte[] {1, 2, 3});
final SimpleExampleMessageData out = new SimpleExampleMessageData();
out.setProcessId(uuid);
@ -96,7 +96,7 @@ public class SimpleExampleMessageTest {
@Test
public void shouldRoundTripFieldThroughStructWithNullable() {
final UUID uuid = UUID.randomUUID();
final Uuid uuid = Uuid.randomUuid();
final ByteBuffer buf1 = ByteBuffer.wrap(new byte[] {1, 2, 3});
final ByteBuffer buf2 = ByteBuffer.wrap(new byte[] {4, 5, 6});
final SimpleExampleMessageData out = new SimpleExampleMessageData();
@ -118,7 +118,7 @@ public class SimpleExampleMessageTest {
@Test
public void shouldRoundTripFieldThroughBuffer() {
final UUID uuid = UUID.randomUUID();
final Uuid uuid = Uuid.randomUuid();
final ByteBuffer buf = ByteBuffer.wrap(new byte[] {1, 2, 3});
final SimpleExampleMessageData out = new SimpleExampleMessageData();
out.setProcessId(uuid);
@ -141,7 +141,7 @@ public class SimpleExampleMessageTest {
@Test
public void shouldRoundTripFieldThroughBufferWithNullable() {
final UUID uuid = UUID.randomUUID();
final Uuid uuid = Uuid.randomUuid();
final ByteBuffer buf1 = ByteBuffer.wrap(new byte[] {1, 2, 3});
final ByteBuffer buf2 = ByteBuffer.wrap(new byte[] {4, 5, 6});
final SimpleExampleMessageData out = new SimpleExampleMessageData();
@ -167,7 +167,7 @@ public class SimpleExampleMessageTest {
@Test
public void shouldImplementEqualsAndHashCode() {
final UUID uuid = UUID.randomUUID();
final Uuid uuid = Uuid.randomUuid();
final ByteBuffer buf = ByteBuffer.wrap(new byte[] {1, 2, 3});
final SimpleExampleMessageData a = new SimpleExampleMessageData();
a.setProcessId(uuid);
@ -259,13 +259,13 @@ public class SimpleExampleMessageTest {
public void testTaggedUuid() {
testRoundTrip(new SimpleExampleMessageData(),
message -> assertEquals(
UUID.fromString("212d54944a8b4fdf94b388b470beb367"),
Uuid.fromString("212d54944a8b4fdf94b388b470beb367"),
message.taggedUuid()));
testRoundTrip(new SimpleExampleMessageData().
setTaggedUuid(UUID.fromString("0123456789abcdef0123456789abcdef")),
setTaggedUuid(Uuid.fromString("0123456789abcdef0123456789abcdef")),
message -> assertEquals(
UUID.fromString("0123456789abcdef0123456789abcdef"),
Uuid.fromString("0123456789abcdef0123456789abcdef"),
message.taggedUuid()));
}

6
generator/src/main/java/org/apache/kafka/message/FieldSpec.java
View File

@ -361,7 +361,7 @@ public final class FieldSpec {
} else if (type instanceof FieldType.UUIDFieldType) {
headerGenerator.addImport(MessageGenerator.UUID_CLASS);
if (fieldDefault.isEmpty()) {
return "UUID.ZERO_UUID";
return "Uuid.ZERO_UUID";
} else {
try {
ByteBuffer uuidBytes = ByteBuffer.wrap(Base64.getUrlDecoder().decode(fieldDefault));
@ -372,7 +372,7 @@ public final class FieldSpec {
name + ": " + fieldDefault, e);
}
headerGenerator.addImport(MessageGenerator.UUID_CLASS);
return "UUID.fromString(\"" + fieldDefault + "\")";
return "Uuid.fromString(\"" + fieldDefault + "\")";
}
} else if (type instanceof FieldType.Float64FieldType) {
if (fieldDefault.isEmpty()) {
@ -463,7 +463,7 @@ public final class FieldSpec {
return "long";
} else if (type instanceof FieldType.UUIDFieldType) {
headerGenerator.addImport(MessageGenerator.UUID_CLASS);
return "UUID";
return "Uuid";
} else if (type instanceof FieldType.Float64FieldType) {
return "double";
} else if (type.isString()) {

2
generator/src/main/java/org/apache/kafka/message/FieldType.java
View File

@ -129,7 +129,7 @@ public interface FieldType {
@Override
public String getBoxedJavaType(HeaderGenerator headerGenerator) {
headerGenerator.addImport(MessageGenerator.UUID_CLASS);
return "UUID";
return "Uuid";
}
@Override

2
generator/src/main/java/org/apache/kafka/message/JsonConverterGenerator.java
View File

@ -177,7 +177,7 @@ public final class JsonConverterGenerator implements MessageClassGenerator {
buffer.printf("}%n");
headerGenerator.addImport(MessageGenerator.UUID_CLASS);
buffer.printf("%s;%n", target.assignmentStatement(String.format(
"UUID.fromString(%s.asText())", target.sourceVariable())));
"Uuid.fromString(%s.asText())", target.sourceVariable())));
} else if (target.field().type() instanceof FieldType.Float64FieldType) {
headerGenerator.addImport(MessageGenerator.MESSAGE_UTIL_CLASS);
buffer.printf("%s;%n", target.assignmentStatement(

6
generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java
View File

@ -559,7 +559,7 @@ public final class MessageDataGenerator implements MessageClassGenerator {
} else if (type instanceof FieldType.Int64FieldType) {
return "_readable.readLong()";
} else if (type instanceof FieldType.UUIDFieldType) {
return "_readable.readUUID()";
return "_readable.readUuid()";
} else if (type instanceof FieldType.Float64FieldType) {
return "_readable.readDouble()";
} else if (type.isStruct()) {
@ -831,7 +831,7 @@ public final class MessageDataGenerator implements MessageClassGenerator {
} else if (type instanceof FieldType.Int64FieldType) {
return String.format("struct.getLong(\"%s\")", name);
} else if (type instanceof FieldType.UUIDFieldType) {
return String.format("struct.getUUID(\"%s\")", name);
return String.format("struct.getUuid(\"%s\")", name);
} else if (type instanceof FieldType.Float64FieldType) {
return String.format("struct.getDouble(\"%s\")", name);
} else if (type.isString()) {
@ -1044,7 +1044,7 @@ public final class MessageDataGenerator implements MessageClassGenerator {
} else if (type instanceof FieldType.Int64FieldType) {
return String.format("_writable.writeLong(%s)", name);
} else if (type instanceof FieldType.UUIDFieldType) {
return String.format("_writable.writeUUID(%s)", name);
return String.format("_writable.writeUuid(%s)", name);
} else if (type instanceof FieldType.Float64FieldType) {
return String.format("_writable.writeDouble(%s)", name);
} else if (type instanceof FieldType.StructType) {

2
generator/src/main/java/org/apache/kafka/message/MessageGenerator.java
View File

@ -94,7 +94,7 @@ public final class MessageGenerator {
static final String BYTES_CLASS = "org.apache.kafka.common.utils.Bytes";
static final String UUID_CLASS = "org.apache.kafka.common.UUID";
static final String UUID_CLASS = "org.apache.kafka.common.Uuid";
static final String BASE_RECORDS_CLASS = "org.apache.kafka.common.record.BaseRecords";

4
streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfo.java
View File

@ -21,6 +21,8 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.protocol.ByteBufferAccessor;
import org.apache.kafka.common.protocol.ObjectSerializationCache;
import org.apache.kafka.streams.errors.TaskAssignmentException;
@ -87,7 +89,7 @@ public class SubscriptionInfo {
validateVersions(version, latestSupportedVersion);
final SubscriptionInfoData data = new SubscriptionInfoData();
data.setVersion(version);
data.setProcessId(new org.apache.kafka.common.UUID(processId.getMostSignificantBits(),
data.setProcessId(new Uuid(processId.getMostSignificantBits(),
processId.getLeastSignificantBits()));
if (version >= 2) {