test_class.declarator: distributionChannel
test_class.modifier: private
test_class.original_string: private DistributionChannel distributionChannel;
test_class.type: DistributionChannel
test_class.var_name: distributionChannel
test_class.file: shardingsphere-scaling/shardingsphere-scaling-core/src/test/java/org/apache/shardingsphere/scaling/core/execute/executor/channel/DistributionChannelTest.java
test_class.identifier: DistributionChannelTest
test_case.body: @Test
      public void assertBroadcastFinishedRecord() throws InterruptedException {
          distributionChannel = new DistributionChannel(2, records -> assertThat(records.size(), is(2)));
          distributionChannel.pushRecord(new FinishedRecord(new NopPosition()));
      }
test_case.class_method_signature: DistributionChannelTest.assertBroadcastFinishedRecord()
test_case.full_signature: @Test public void assertBroadcastFinishedRecord()
test_case.identifier: assertBroadcastFinishedRecord
test_case.invocations: assertThat size is pushRecord
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void assertBroadcastFinishedRecord()
focal_class.declarator: channelNumber
focal_class.modifier: private final
focal_class.original_string: private final int channelNumber;
focal_class.type: int
focal_class.var_name: channelNumber
focal_class.declarator: channels = new HashMap<>()
focal_class.modifier: private final
focal_class.original_string: private final Map<String, MemoryChannel> channels = new HashMap<>();
focal_class.type: Map<String, MemoryChannel>
focal_class.var_name: channels
focal_class.declarator: channelAssignment = new HashMap<>()
focal_class.modifier: private final
focal_class.original_string: private final Map<String, String> channelAssignment = new HashMap<>();
focal_class.type: Map<String, String>
focal_class.var_name: channelAssignment
focal_class.declarator: ackCallback
focal_class.modifier: private final
focal_class.original_string: private final AckCallback ackCallback;
focal_class.type: AckCallback
focal_class.var_name: ackCallback
focal_class.declarator: toBeAcknowledgeRecords = new ConcurrentLinkedQueue<>()
focal_class.modifier: private final
focal_class.original_string: private final Queue<Record> toBeAcknowledgeRecords = new ConcurrentLinkedQueue<>();
focal_class.type: Queue<Record>
focal_class.var_name: toBeAcknowledgeRecords
focal_class.declarator: pendingAcknowledgeRecords = new ConcurrentHashMap<>()
focal_class.modifier: private final
focal_class.original_string: private final Map<Position, Record> pendingAcknowledgeRecords = new ConcurrentHashMap<>();
focal_class.type: Map<Position, Record>
focal_class.var_name: pendingAcknowledgeRecords
focal_class.declarator: scheduleAckRecordsExecutor
focal_class.modifier: private
focal_class.original_string: private ScheduledExecutorService scheduleAckRecordsExecutor;
focal_class.type: ScheduledExecutorService
focal_class.var_name: scheduleAckRecordsExecutor
focal_class.file: shardingsphere-scaling/shardingsphere-scaling-core/src/main/java/org/apache/shardingsphere/scaling/core/execute/executor/channel/DistributionChannel.java
focal_class.identifier: DistributionChannel
focal_class.interfaces: implements Channel
focal_class.class_method_signature: DistributionChannel.DistributionChannel(final int channelNumber, final AckCallback ackCallback)
focal_class.full_signature: public  DistributionChannel(final int channelNumber, final AckCallback ackCallback)
focal_class.identifier: DistributionChannel
focal_class.modifiers: public
focal_class.parameters: (final int channelNumber, final AckCallback ackCallback)
focal_class.signature:  DistributionChannel(final int channelNumber, final AckCallback ackCallback)
focal_class.class_method_signature: DistributionChannel.scheduleAckRecords()
focal_class.full_signature: private void scheduleAckRecords()
focal_class.identifier: scheduleAckRecords
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void scheduleAckRecords()
focal_class.class_method_signature: DistributionChannel.ackRecords0()
focal_class.full_signature: private synchronized void ackRecords0()
focal_class.identifier: ackRecords0
focal_class.modifiers: private synchronized
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void ackRecords0()
focal_class.class_method_signature: DistributionChannel.pushRecord(final Record record)
focal_class.full_signature: @Override public void pushRecord(final Record record)
focal_class.identifier: pushRecord
focal_class.modifiers: @Override public
focal_class.parameters: (final Record record)
focal_class.return: void
focal_class.signature: void pushRecord(final Record record)
focal_class.class_method_signature: DistributionChannel.fetchRecords(final int batchSize, final int timeout)
focal_class.full_signature: @Override public List<Record> fetchRecords(final int batchSize, final int timeout)
focal_class.identifier: fetchRecords
focal_class.modifiers: @Override public
focal_class.parameters: (final int batchSize, final int timeout)
focal_class.return: List<Record>
focal_class.signature: List<Record> fetchRecords(final int batchSize, final int timeout)
focal_class.class_method_signature: DistributionChannel.ack()
focal_class.full_signature: @Override public void ack()
focal_class.identifier: ack
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void ack()
focal_class.class_method_signature: DistributionChannel.close()
focal_class.full_signature: @Override public void close()
focal_class.identifier: close
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void close()
focal_class.class_method_signature: DistributionChannel.findChannel()
focal_class.full_signature: private Channel findChannel()
focal_class.identifier: findChannel
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: Channel
focal_class.signature: Channel findChannel()
focal_class.class_method_signature: DistributionChannel.checkAssignment(final String threadId)
focal_class.full_signature: private void checkAssignment(final String threadId)
focal_class.identifier: checkAssignment
focal_class.modifiers: private
focal_class.parameters: (final String threadId)
focal_class.return: void
focal_class.signature: void checkAssignment(final String threadId)
focal_class.class_method_signature: DistributionChannel.assignmentChannel(final String threadId)
focal_class.full_signature: private void assignmentChannel(final String threadId)
focal_class.identifier: assignmentChannel
focal_class.modifiers: private
focal_class.parameters: (final String threadId)
focal_class.return: void
focal_class.signature: void assignmentChannel(final String threadId)
focal_method.body: @Override
      public void pushRecord(final Record record) throws InterruptedException {
          if (FinishedRecord.class.equals(record.getClass())) {
              // broadcast
              for (Entry<String, MemoryChannel> entry : channels.entrySet()) {
                  entry.getValue().pushRecord(record);
              }
          } else if (DataRecord.class.equals(record.getClass())) {
              toBeAcknowledgeRecords.add(record);
              // hash by table name
              DataRecord dataRecord = (DataRecord) record;
              String index = Integer.toString(Math.abs(dataRecord.hashCode()) % channelNumber);
              channels.get(index).pushRecord(dataRecord);
          } else if (PlaceholderRecord.class.equals(record.getClass())) {
              toBeAcknowledgeRecords.add(record);
              pendingAcknowledgeRecords.put(record.getPosition(), record);
          } else {
              throw new RuntimeException("Not Support Record Type");
          }
      }
focal_method.class_method_signature: DistributionChannel.pushRecord(final Record record)
focal_method.full_signature: @Override public void pushRecord(final Record record)
focal_method.identifier: pushRecord
focal_method.invocations: equals getClass entrySet pushRecord getValue equals getClass add toString abs hashCode pushRecord get equals getClass add put getPosition
focal_method.modifiers: @Override public
focal_method.parameters: (final Record record)
focal_method.return: void
focal_method.signature: void pushRecord(final Record record)