Skip to content
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

Fixes for NoOpTransportProvider #940

Merged
merged 7 commits into from
Jun 13, 2023
Merged

Fixes for NoOpTransportProvider #940

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
3911ef0
Skipping destination connection string checks for NoOp Transport Prov…
jzakaryan May 15, 2023
ed2634d
Version bumo
jzakaryan May 15, 2023
d169ed2
Workaround for copying destination for NoOp TP
jzakaryan May 16, 2023
8611248
Version bump
jzakaryan May 16, 2023
6f2621b
Merge branch 'master' into jzakarya/noop-transport-fixes
jzakaryan Jun 12, 2023
3b068d3
Minor improvements
jzakaryan Jun 12, 2023
42552fd
Minor improvements
jzakaryan Jun 12, 2023
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
13 changes: 8 additions & 5 deletions ...a/com/linkedin/datastream/connectors/kafka/mirrormaker/KafkaMirrorMakerConnectorTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,8 +86,8 @@ public class KafkaMirrorMakerConnectorTask extends AbstractKafkaBasedConnectorTa
private static final String KAFKA_ORIGIN_OFFSET = "kafka-origin-offset";
private static final Duration LOCK_ACQUIRE_TIMEOUT = Duration.ofMinutes(3);
private static final String TASK_LOCK_ACQUIRE_ERROR_RATE = "taskLockAcquireErrorRate";

private static final String DATASTREAM_NAME_BASED_CLIENT_ID_FORMAT = "%s-%s";
private static final String NOOP_TRANSPORT_PROVIDER_NAME = "NoOp";

// constants for flushless mode and flow control
protected static final String CONFIG_MAX_IN_FLIGHT_MSGS_THRESHOLD = "maxInFlightMessagesThreshold";
Expand Down Expand Up @@ -261,10 +261,13 @@ protected DatastreamProducerRecord translate(ConsumerRecord<?, ?> fromKafka, Ins
builder.addEvent(envelope);
builder.setEventsSourceTimestamp(eventsSourceTimestamp);
builder.setSourceCheckpoint(new KafkaMirrorMakerCheckpoint(topic, partition, offset).toString());
builder.setDestination(_datastreamTask.getDatastreamDestination()
.getConnectionString()
.replace(KafkaMirrorMakerConnector.MM_TOPIC_PLACEHOLDER,
StringUtils.isBlank(_destinationTopicPrefix) ? topic : _destinationTopicPrefix + topic));

if (!_datastreamTask.getTransportProviderName().equalsIgnoreCase(NOOP_TRANSPORT_PROVIDER_NAME)) {
builder.setDestination(_datastreamTask.getDatastreamDestination()
.getConnectionString()
.replace(KafkaMirrorMakerConnector.MM_TOPIC_PLACEHOLDER,
StringUtils.isBlank(_destinationTopicPrefix) ? topic : _destinationTopicPrefix + topic));
}
if (_isIdentityMirroringEnabled) {
builder.setPartition(partition);
}
Expand Down
8 changes: 6 additions & 2 deletions datastream-server/src/main/java/com/linkedin/datastream/server/EventProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ public class EventProducer implements DatastreamEventProducer {
static final String EVENTS_SEND_LATENCY_MS_STRING = "eventsSendLatencyMs";
static final String THROUGHPUT_VIOLATING_EVENTS_LATENCY_MS_STRING = "throughputViolatingEventsLatencyMs";
static final String THROUGHPUT_VIOLATING_EVENTS_SEND_LATENCY_MS_STRING = "throughputViolatingEventsSendLatencyMs";
static final String NOOP_TRANSPORT_PROVIDER_NAME = "NoOp";
jzakaryan marked this conversation as resolved.
Show resolved Hide resolved

private static final String MODULE = EventProducer.class.getSimpleName();
private static final String METRICS_PREFIX = MODULE + MetricsAware.KEY_REGEX;
Expand Down Expand Up @@ -271,8 +272,11 @@ private DatastreamRecordMetadata helperSendOrBroadcast(DatastreamProducerRecord

try {
// Send the event to the transport
String destination =
record.getDestination().orElse(_datastreamTask.getDatastreamDestination().getConnectionString());
String destination = "dummy";
jzakaryan marked this conversation as resolved.
Show resolved Hide resolved
if (!_datastreamTask.getTransportProviderName().equalsIgnoreCase(NOOP_TRANSPORT_PROVIDER_NAME)) {
destination =
record.getDestination().orElse(_datastreamTask.getDatastreamDestination().getConnectionString());
}
record.setEventsSendTimestamp(System.currentTimeMillis());
long recordEventsSourceTimestamp = record.getEventsSourceTimestamp();
long recordEventsSendTimestamp = record.getEventsSendTimestamp().orElse(0L);
Expand Down
2 changes: 1 addition & 1 deletion gradle/maven.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
allprojects {
version = "5.3.5-SNAPSHOT"
version = "5.3.6-SNAPSHOT"
}

subprojects {
Expand Down