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

Jump to bottom

PD-220325-Add-logs-for-the-megabus-for-debugging-purpose #785

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

PD-220325-Add-logs-for-the-megabus-for-debugging-purpose #785

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
cc96faa
PD-220325: megabus logs for provisioning issue
Abhijeet-BazaarVoice Oct 9, 2023
852b564
PD-220325: added class to log
Abhijeet-BazaarVoice Oct 9, 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
5 changes: 5 additions & 0 deletions megabus/src/main/java/com/bazaarvoice/megabus/refproducer/MegabusRefProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,8 @@ public class MegabusRefProducer extends AbstractScheduledService {
private final Meter _eventMeter;
private final Meter _errorMeter;

private static final Logger logger = LoggerFactory.getLogger(MegabusRefProducer.class);
Abhijeet-BazaarVoice marked this conversation as resolved.
Show resolved Hide resolved

public MegabusRefProducer(MegabusRefProducerConfiguration config, DatabusEventStore eventStore,
RateLimitedLogFactory logFactory, MetricRegistry metricRegistry,
Producer<String, JsonNode> producer, ObjectMapper objectMapper, Topic topic,
Expand Down Expand Up @@ -147,6 +149,9 @@ boolean peekAndAckEvents() {
.map(ref -> new MegabusRef(ref.getTable(), ref.getKey(), ref.getChangeId(), _clock.instant(), MegabusRef.RefType.NORMAL))
.collect(Collectors.groupingBy(ref -> {
String key = Coordinate.of(ref.getTable(), ref.getKey()).toString();
if(ref.getTable().contains("apikey")){
logger.info("debugging mega-bus delay while provisioning apikey 2: = {}",key);
Abhijeet-BazaarVoice marked this conversation as resolved.
Show resolved Hide resolved
Abhijeet-BazaarVoice marked this conversation as resolved.
Show resolved Hide resolved
}
return Utils.toPositive(Utils.murmur2(key.getBytes())) % _topic.getPartitions();
}, Collectors.toList()))
.entrySet()
Expand Down
6 changes: 4 additions & 2 deletions megabus/src/main/java/com/bazaarvoice/megabus/tableevents/TableEventProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ public class TableEventProcessor extends AbstractScheduledService {

private static final int FUTURE_BATCH_SIZE = 10000;

private static final Logger _log = LoggerFactory.getLogger(TableEventProcessor.class);
private static final Logger logger = LoggerFactory.getLogger(TableEventProcessor.class);

private final TableEventRegistry _tableEventRegistry;
private final MetricRegistry _metricRegistry;
Expand Down Expand Up @@ -80,11 +80,13 @@ protected void runOneIteration() throws Exception {
}

private void processTableEvent(String table, String uuid, MegabusRef.RefType refType) {

Iterator<Future<RecordMetadata>> futures = _tableEventTools.getIdsForStorage(table, uuid)
.map(key -> new MegabusRef(table, key, TimeUUIDs.minimumUuid(), null, refType))
.map(ref -> {
String key = Coordinate.of(ref.getTable(), ref.getKey()).toString();
if(table.contains("apikey")){
logger.info("debugging mega-bus delay while provisioning apikey 1: = {}",key);
}
return new ProducerRecord<String, JsonNode>(_topic.getName(),
Utils.toPositive(Utils.murmur2(key.getBytes())) % _topic.getPartitions(),
TimeUUIDs.newUUID().toString(),_objectMapper.valueToTree(Collections.singletonList(ref)));
Expand Down