Skip to content

Use switch expressions introduced in Java 14 to simplify code #18371

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

Closed
ijuma wants to merge 1 commit into from
Closed

Use switch expressions introduced in Java 14 to simplify code #18371

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
17 changes: 4 additions & 13 deletions connect/api/src/main/java/org/apache/kafka/connect/data/Schema.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,19 +117,10 @@ public String getName() {
}

public boolean isPrimitive() {
switch (this) {
case INT8:
case INT16:
case INT32:
case INT64:
case FLOAT32:
case FLOAT64:
case BOOLEAN:
case STRING:
case BYTES:
return true;
}
return false;
return switch (this) {
case INT8, INT16, INT32, INT64, FLOAT32, FLOAT64, BOOLEAN, STRING, BYTES -> true;
default -> false;
};
}
}

Expand Down
63 changes: 18 additions & 45 deletions connect/api/src/main/java/org/apache/kafka/connect/data/SchemaProjector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import java.util.Objects;
import java.util.Set;


/**
* <p>
* SchemaProjector is a utility to project a value between compatible schemas and throw exceptions
Expand Down Expand Up @@ -78,25 +79,13 @@ public static Object project(Schema source, Object record, Schema target) throws
}

private static Object projectRequiredSchema(Schema source, Object record, Schema target) throws SchemaProjectorException {
switch (target.type()) {
case INT8:
case INT16:
case INT32:
case INT64:
case FLOAT32:
case FLOAT64:
case BOOLEAN:
case BYTES:
case STRING:
return projectPrimitive(source, record, target);
case STRUCT:
return projectStruct(source, (Struct) record, target);
case ARRAY:
return projectArray(source, record, target);
case MAP:
return projectMap(source, record, target);
}
return null;

@clolov clolov Jan 6, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

In other places you have translated this to default -> null in the new syntax. Shouldn't this be the same or am I missing something obvious?

@ijuma ijuma Jan 7, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

This was directly translated by IntelliJ - I started (but haven't completed) the review of every case to make sure it makes sense. Generally speaking, if the switch is exhaustive, then it doesn't need to return null if we're ok with throwing an exception if a new enum element is added later and the code is not updated.

return switch (target.type()) {
case INT8, INT16, INT32, INT64, FLOAT32, FLOAT64, BOOLEAN, BYTES, STRING ->
projectPrimitive(source, record, target);
case STRUCT -> projectStruct(source, (Struct) record, target);
case ARRAY -> projectArray(source, record, target);
case MAP -> projectMap(source, record, target);
};
}

private static Object projectStruct(Schema source, Struct sourceStruct, Schema target) throws SchemaProjectorException {
Expand Down Expand Up @@ -159,35 +148,19 @@ private static Object projectMap(Schema source, Object record, Schema target) th
private static Object projectPrimitive(Schema source, Object record, Schema target) throws SchemaProjectorException {
assert source.type().isPrimitive();
assert target.type().isPrimitive();
Object result;
if (isPromotable(source.type(), target.type()) && record instanceof Number) {
Number numberRecord = (Number) record;
switch (target.type()) {
case INT8:
result = numberRecord.byteValue();
break;
case INT16:
result = numberRecord.shortValue();
break;
case INT32:
result = numberRecord.intValue();
break;
case INT64:
result = numberRecord.longValue();
break;
case FLOAT32:
result = numberRecord.floatValue();
break;
case FLOAT64:
result = numberRecord.doubleValue();
break;
default:
throw new SchemaProjectorException("Not promotable type.");
}
} else {
result = record;
return switch (target.type()) {
case INT8 -> numberRecord.byteValue();
case INT16 -> numberRecord.shortValue();
case INT32 -> numberRecord.intValue();
case INT64 -> numberRecord.longValue();
case FLOAT32 -> numberRecord.floatValue();
case FLOAT64 -> numberRecord.doubleValue();
default -> throw new SchemaProjectorException("Not promotable type.");
};
}
return result;
return record;
}

private static boolean isPromotable(Type sourceType, Type targetType) {
Expand Down
65 changes: 23 additions & 42 deletions connect/api/src/main/java/org/apache/kafka/connect/data/Values.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -431,33 +431,20 @@ protected static Object convertTo(Schema toSchema, Schema fromSchema, Object val
}
throw new DataException("Unable to convert a null value to a schema that requires a value");
}
switch (toSchema.type()) {
case BYTES:
return convertMaybeLogicalBytes(toSchema, value);
case STRING:
return convertToString(fromSchema, value);
case BOOLEAN:
return convertToBoolean(fromSchema, value);
case INT8:
return convertToByte(fromSchema, value);
case INT16:
return convertToShort(fromSchema, value);
case INT32:
return convertMaybeLogicalInteger(toSchema, fromSchema, value);
case INT64:
return convertMaybeLogicalLong(toSchema, fromSchema, value);
case FLOAT32:
return convertToFloat(fromSchema, value);
case FLOAT64:
return convertToDouble(fromSchema, value);
case ARRAY:
return convertToArray(toSchema, value);
case MAP:
return convertToMapInternal(toSchema, value);
case STRUCT:
return convertToStructInternal(toSchema, value);
}
throw new DataException("Unable to convert " + value + " (" + value.getClass() + ") to " + toSchema);

@clolov clolov Jan 6, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

For my curiosity, will the exception here not change with the new syntax? If it changes, do upper layers handle it correctly?

@ijuma ijuma Jan 7, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

This case is never hit since the switch is exhaustive.

return switch (toSchema.type()) {
case BYTES -> convertMaybeLogicalBytes(toSchema, value);
case STRING -> convertToString(fromSchema, value);
case BOOLEAN -> convertToBoolean(fromSchema, value);
case INT8 -> convertToByte(fromSchema, value);
case INT16 -> convertToShort(fromSchema, value);
case INT32 -> convertMaybeLogicalInteger(toSchema, fromSchema, value);
case INT64 -> convertMaybeLogicalLong(toSchema, fromSchema, value);
case FLOAT32 -> convertToFloat(fromSchema, value);
case FLOAT64 -> convertToDouble(fromSchema, value);
case ARRAY -> convertToArray(toSchema, value);
case MAP -> convertToMapInternal(toSchema, value);
case STRUCT -> convertToStructInternal(toSchema, value);
};
}

private static Serializable convertMaybeLogicalBytes(Schema toSchema, Object value) {
Expand Down Expand Up @@ -1144,21 +1131,15 @@ private static Schema mergeSchemas(Schema previous, Schema newSchema) {
Type previousType = previous.type();
Type newType = newSchema.type();
if (previousType != newType) {
switch (previous.type()) {
case INT8:
return commonSchemaForInt8(newSchema, newType);
case INT16:
return commonSchemaForInt16(previous, newSchema, newType);
case INT32:
return commonSchemaForInt32(previous, newSchema, newType);
case INT64:
return commonSchemaForInt64(previous, newSchema, newType);
case FLOAT32:
return commonSchemaForFloat32(previous, newSchema, newType);
case FLOAT64:
return commonSchemaForFloat64(previous, newType);
}
return null;
return switch (previous.type()) {
case INT8 -> commonSchemaForInt8(newSchema, newType);
case INT16 -> commonSchemaForInt16(previous, newSchema, newType);
case INT32 -> commonSchemaForInt32(previous, newSchema, newType);
case INT64 -> commonSchemaForInt64(previous, newSchema, newType);
case FLOAT32 -> commonSchemaForFloat32(previous, newSchema, newType);
case FLOAT64 -> commonSchemaForFloat64(previous, newType);
default -> null;
};
}
if (previous.isOptional() == newSchema.isOptional()) {
// Use the optional one
Expand Down
14 changes: 6 additions & 8 deletions connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,14 +153,12 @@ public JsonNode toJson(final Schema schema, final Object value, final JsonConver
throw new DataException("Invalid type for Decimal, expected BigDecimal but was " + value.getClass());

final BigDecimal decimal = (BigDecimal) value;
switch (config.decimalFormat()) {
case NUMERIC:
return JSON_NODE_FACTORY.numberNode(decimal);
case BASE64:
return JSON_NODE_FACTORY.binaryNode(Decimal.fromLogical(schema, decimal));
default:
throw new DataException("Unexpected " + JsonConverterConfig.DECIMAL_FORMAT_CONFIG + ": " + config.decimalFormat());
}
return switch (config.decimalFormat()) {
case NUMERIC -> JSON_NODE_FACTORY.numberNode(decimal);
case BASE64 -> JSON_NODE_FACTORY.binaryNode(Decimal.fromLogical(schema, decimal));
default ->
throw new DataException("Unexpected " + JsonConverterConfig.DECIMAL_FORMAT_CONFIG + ": " + config.decimalFormat());
};
}

@Override
Expand Down
42 changes: 20 additions & 22 deletions ...t/runtime/src/main/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -424,24 +424,22 @@ private TransactionBoundaryManager buildTransactionManager(
SourceConnectorConfig sourceConfig,
WorkerTransactionContext transactionContext) {
TransactionBoundary boundary = sourceConfig.transactionBoundary();
switch (boundary) {
case POLL:
return new TransactionBoundaryManager() {
@Override
protected boolean shouldCommitTransactionForBatch(long currentTimeMs) {
return true;
}

@Override
protected boolean shouldCommitFinalTransaction() {
return true;
}
};
return switch (boundary) {
case POLL -> new TransactionBoundaryManager() {
@Override
protected boolean shouldCommitTransactionForBatch(long currentTimeMs) {
return true;
}

case INTERVAL:
@Override
protected boolean shouldCommitFinalTransaction() {
return true;
}
};
case INTERVAL -> {
long transactionBoundaryInterval = Optional.ofNullable(sourceConfig.transactionBoundaryInterval())
.orElse(workerConfig.offsetCommitInterval());
return new TransactionBoundaryManager() {
yield new TransactionBoundaryManager() {
private final long commitInterval = transactionBoundaryInterval;
private long lastCommit;

Expand All @@ -461,14 +459,14 @@ protected boolean shouldCommitTransactionForBatch(long currentTimeMs) {
}

@Override
protected boolean shouldCommitFinalTransaction() {
protected boolean shouldCommitFinalTransaction() {
return true;
}
};

case CONNECTOR:
}
case CONNECTOR -> {
Objects.requireNonNull(transactionContext, "Transaction context must be provided when using connector-defined transaction boundaries");
return new TransactionBoundaryManager() {
yield new TransactionBoundaryManager() {
@Override
protected boolean shouldCommitFinalTransaction() {
return shouldCommitTransactionForBatch(time.milliseconds());
Expand Down Expand Up @@ -509,9 +507,9 @@ private void maybeAbortTransaction() {
transactionOpen = false;
}
};
default:
throw new IllegalArgumentException("Unrecognized transaction boundary: " + boundary);
}
}
default -> throw new IllegalArgumentException("Unrecognized transaction boundary: " + boundary);
};
}

TransactionMetricsGroup transactionMetricsGroup() {
Expand Down
16 changes: 6 additions & 10 deletions .../main/java/org/apache/kafka/connect/runtime/distributed/ConnectProtocolCompatibility.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,16 +99,12 @@ public static ConnectProtocolCompatibility compatibility(String name) {
* @return the enum that corresponds to the protocol compatibility mode
*/
public static ConnectProtocolCompatibility fromProtocolVersion(short protocolVersion) {
switch (protocolVersion) {
case CONNECT_PROTOCOL_V0:
return EAGER;
case CONNECT_PROTOCOL_V1:
return COMPATIBLE;
case CONNECT_PROTOCOL_V2:
return SESSIONED;
default:
throw new IllegalArgumentException("Unknown Connect protocol version: " + protocolVersion);
}
return switch (protocolVersion) {
case CONNECT_PROTOCOL_V0 -> EAGER;
case CONNECT_PROTOCOL_V1 -> COMPATIBLE;
case CONNECT_PROTOCOL_V2 -> SESSIONED;
default -> throw new IllegalArgumentException("Unknown Connect protocol version: " + protocolVersion);
};
}

@Override
Expand Down
17 changes: 7 additions & 10 deletions ...runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,16 +183,13 @@ public JoinGroupRequestProtocolCollection metadata() {
configSnapshot = configStorage.snapshot();
final ExtendedAssignment localAssignmentSnapshot = assignmentSnapshot;
ExtendedWorkerState workerState = new ExtendedWorkerState(restUrl, configSnapshot.offset(), localAssignmentSnapshot);
switch (protocolCompatibility) {
case EAGER:
return ConnectProtocol.metadataRequest(workerState);
case COMPATIBLE:
return IncrementalCooperativeConnectProtocol.metadataRequest(workerState, false);
case SESSIONED:
return IncrementalCooperativeConnectProtocol.metadataRequest(workerState, true);
default:
throw new IllegalStateException("Unknown Connect protocol compatibility mode " + protocolCompatibility);
}
return switch (protocolCompatibility) {
case EAGER -> ConnectProtocol.metadataRequest(workerState);
case COMPATIBLE -> IncrementalCooperativeConnectProtocol.metadataRequest(workerState, false);
case SESSIONED -> IncrementalCooperativeConnectProtocol.metadataRequest(workerState, true);
default ->
throw new IllegalStateException("Unknown Connect protocol compatibility mode " + protocolCompatibility);
};
}

@Override
Expand Down
11 changes: 4 additions & 7 deletions connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -410,13 +410,10 @@ public static String simpleName(PluginDesc<?> plugin) {
*/
public static String prunedName(PluginDesc<?> plugin) {
// It's currently simpler to switch on type than do pattern matching.
switch (plugin.type()) {
case SOURCE:
case SINK:
return prunePluginName(plugin, "Connector");
default:
return prunePluginName(plugin, plugin.type().simpleName());
}
return switch (plugin.type()) {
case SOURCE, SINK -> prunePluginName(plugin, "Connector");
default -> prunePluginName(plugin, plugin.type().simpleName());
};
}

private static String prunePluginName(PluginDesc<?> plugin, String suffix) {
Expand Down
16 changes: 6 additions & 10 deletions .../src/main/java/org/apache/kafka/connect/runtime/rest/entities/CreateConnectorRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,16 +87,12 @@ public static InitialState forValue(String value) {
}

public TargetState toTargetState() {
switch (this) {
case RUNNING:
return TargetState.STARTED;
case PAUSED:
return TargetState.PAUSED;
case STOPPED:
return TargetState.STOPPED;
default:
throw new IllegalArgumentException("Unknown initial state: " + this);
}
return switch (this) {
case RUNNING -> TargetState.STARTED;
case PAUSED -> TargetState.PAUSED;
case STOPPED -> TargetState.STOPPED;
default -> throw new IllegalArgumentException("Unknown initial state: " + this);
};
}
}
}
Loading