HBASE-24718 : Generic NamedQueue framework for multiple use-cases (Refactor SlowLog responses) #2052
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,101 @@ | ||
| /* | ||
| * | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.hadoop.hbase.namequeues; | ||
|
|
||
| import com.lmax.disruptor.EventHandler; | ||
| import com.lmax.disruptor.RingBuffer; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
|
|
||
| import org.apache.hadoop.conf.Configuration; | ||
| import org.apache.hadoop.hbase.namequeues.request.NamedQueueGetRequest; | ||
| import org.apache.hadoop.hbase.namequeues.response.NamedQueueGetResponse; | ||
| import org.apache.yetus.audience.InterfaceAudience; | ||
|
|
||
| /** | ||
| * Event Handler run by disruptor ringbuffer consumer. | ||
| * Although this is generic implementation for namedQueue, it can have individual queue specific | ||
| * logic. | ||
| */ | ||
| @InterfaceAudience.Private | ||
| class LogEventHandler implements EventHandler<RingBufferEnvelope> { | ||
|
|
||
| // Map that binds namedQueues to corresponding queue service implementation. | ||
| // If NamedQueue of specific type is enabled, corresponding service will be used to | ||
| // insert and retrieve records. | ||
| // Individual queue sizes should be determined based on their individual configs within | ||
| // each service. | ||
| private final Map<NamedQueuePayload.NamedQueueEvent, NamedQueueService> namedQueueServices = | ||
| new HashMap<>(); | ||
|
|
||
| LogEventHandler(final Configuration conf) { | ||
| // add all service mappings here | ||
| namedQueueServices | ||
|
||
| .put(NamedQueuePayload.NamedQueueEvent.SLOW_LOG, new SlowLogQueueService(conf)); | ||
| } | ||
|
|
||
| /** | ||
| * Called when a publisher has published an event to the {@link RingBuffer}. | ||
| * This is generic consumer of disruptor ringbuffer and for each new namedQueue that we | ||
| * add, we should also provide specific consumer logic here. | ||
| * | ||
| * @param event published to the {@link RingBuffer} | ||
| * @param sequence of the event being processed | ||
| * @param endOfBatch flag to indicate if this is the last event in a batch from | ||
| * the {@link RingBuffer} | ||
| */ | ||
| @Override | ||
| public void onEvent(RingBufferEnvelope event, long sequence, boolean endOfBatch) { | ||
| final NamedQueuePayload namedQueuePayload = event.getPayload(); | ||
| // consume ringbuffer payload based on event type | ||
| namedQueueServices.get(namedQueuePayload.getNamedQueueEvent()) | ||
| .consumeEventFromDisruptor(namedQueuePayload); | ||
| } | ||
|
|
||
| /** | ||
| * Cleans up queues maintained by services. | ||
| * | ||
| * @param namedQueueEvent type of queue to clear | ||
| * @return true if queue is cleaned up, false otherwise | ||
| */ | ||
| boolean clearNamedQueue(NamedQueuePayload.NamedQueueEvent namedQueueEvent) { | ||
| return namedQueueServices.get(namedQueueEvent).clearNamedQueue(); | ||
| } | ||
|
|
||
| /** | ||
| * Add all in memory queue records to system table. The implementors can use system table | ||
| * or direct HDFS file or ZK as persistence system. | ||
| */ | ||
| void persistAll(NamedQueuePayload.NamedQueueEvent namedQueueEvent) { | ||
| namedQueueServices.get(namedQueueEvent).persistAll(); | ||
| } | ||
|
|
||
| /** | ||
| * Retrieve in memory queue records from ringbuffer | ||
| * | ||
| * @param request namedQueue request with event type | ||
| * @return queue records from ringbuffer after filter (if applied) | ||
| */ | ||
| NamedQueueGetResponse getNamedQueueRecords(NamedQueueGetRequest request) { | ||
| return namedQueueServices.get(request.getNamedQueueEvent()).getNamedQueueRecords(request); | ||
| } | ||
|
|
||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| /* | ||
| * | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.hadoop.hbase.namequeues; | ||
|
|
||
| import org.apache.yetus.audience.InterfaceAudience; | ||
|
|
||
| /** | ||
| * Base payload to be prepared by client to send various namedQueue events for in-memory | ||
| * ring buffer storage in either HMaster or RegionServer. | ||
| * e.g slowLog responses | ||
| */ | ||
| @InterfaceAudience.Private | ||
| public class NamedQueuePayload { | ||
|
|
||
| public enum NamedQueueEvent { | ||
| SLOW_LOG | ||
| } | ||
|
|
||
| private final NamedQueueEvent namedQueueEvent; | ||
|
|
||
| public NamedQueuePayload(NamedQueueEvent namedQueueEvent) { | ||
| if (namedQueueEvent == null) { | ||
| throw new RuntimeException("NamedQueuePayload with null namedQueueEvent"); | ||
| } | ||
| this.namedQueueEvent = namedQueueEvent; | ||
| } | ||
|
|
||
| public NamedQueueEvent getNamedQueueEvent() { | ||
| return namedQueueEvent; | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.