@Test(expected = IllegalArgumentException.class) public void testAddUserMessageValidity3() throws Exception { talosProducer = new TalosProducer(talosProducerConfig, talosResourceName, talosAdminMock, talosClientFactoryMock, partitionSenderMock, new SimpleTopicAbnormalCallback(), new TestCallback()); String bigStr = getRandomString( Constants.TALOS_SINGLE_MESSAGE_BYTES_MAXIMAL + 1); ArrayList<Message> list = new ArrayList<Message>(); list.add(new Message(ByteBuffer.wrap(bigStr.getBytes()))); talosProducer.addUserMessage(list); }
public synchronized void addUserMessage(List<Message> msgList, long timeoutMillis) throws ProducerNotActiveException, AddMessageTimeoutException { if (!isActive()) { throw new ProducerNotActiveException("Producer is not active, " + "current state: " + producerState); } while (bufferedCount.isFull()) { synchronized (globalLock) { try { LOG.info("too many buffered messages, globalLock is active." + " message number: " + bufferedCount.getBufferedMsgNumber() + ", message bytes: " + bufferedCount.getBufferedMsgBytes()); long startWaitTime = System.currentTimeMillis(); globalLock.wait(timeoutMillis); if (System.currentTimeMillis() - startWaitTime >= timeoutMillis) { throw new AddMessageTimeoutException("Producer buffer is full and " + "addUserMessage timeout by: " + timeoutMillis + " millis"); } } catch (InterruptedException e) { LOG.error("addUserMessage global lock wait is interrupt.", e); } } } doAddUserMessage(msgList); }
TalosProducer { public synchronized void addUserMessage(List<Message> msgList, long timeoutMillis) throws ProducerNotActiveException, AddMessageTimeoutException { if (!isActive()) { throw new ProducerNotActiveException("Producer is not active, " + "current state: " + producerState); } while (bufferedCount.isFull()) { synchronized (globalLock) { try { LOG.info("too many buffered messages, globalLock is active." + " message number: " + bufferedCount.getBufferedMsgNumber() + ", message bytes: " + bufferedCount.getBufferedMsgBytes()); long startWaitTime = System.currentTimeMillis(); globalLock.wait(timeoutMillis); if (System.currentTimeMillis() - startWaitTime >= timeoutMillis) { throw new AddMessageTimeoutException("Producer buffer is full and " + "addUserMessage timeout by: " + timeoutMillis + " millis"); } } catch (InterruptedException e) { LOG.error("addUserMessage global lock wait is interrupt.", e); } } } doAddUserMessage(msgList); } }
TalosProducer { public synchronized void addUserMessage(List<Message> msgList, long timeoutMillis) throws ProducerNotActiveException, AddMessageTimeoutException { if (!isActive()) { throw new ProducerNotActiveException("Producer is not active, " + "current state: " + producerState); } while (bufferedCount.isFull()) { synchronized (globalLock) { try { LOG.info("too many buffered messages, globalLock is active." + " message number: " + bufferedCount.getBufferedMsgNumber() + ", message bytes: " + bufferedCount.getBufferedMsgBytes()); long startWaitTime = System.currentTimeMillis(); globalLock.wait(timeoutMillis); if (System.currentTimeMillis() - startWaitTime >= timeoutMillis) { throw new AddMessageTimeoutException("Producer buffer is full and " + "addUserMessage timeout by: " + timeoutMillis + " millis"); } } catch (InterruptedException e) { LOG.error("addUserMessage global lock wait is interrupt.", e); } } } doAddUserMessage(msgList); } TalosProducer(TalosProducerConfig producerConfig,
      TopicTalosResourceName topicTalosResourceName,
      TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback); @Deprecated  TalosProducer(TalosProducerConfig producerConfig, Credential credential,
      TopicTalosResourceName topicTalosResourceName,
      TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback);  TalosProducer(TalosProducerConfig producerConfig, Credential credential,
      String topicName, TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback);  TalosProducer(TalosProducerConfig producerConfig, Credential credential,
      String topicName, Partitioner partitioner,
      TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback); @Deprecated  TalosProducer(TalosProducerConfig producerConfig, Credential credential,
      TopicTalosResourceName topicTalosResourceName, Partitioner partitioner,
      TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback);  TalosProducer(TalosProducerConfig producerConfig,
      TopicTalosResourceName topicTalosResourceName, TalosAdmin talosAdmin,
      TalosClientFactory talosClientFactory,
      PartitionSender partitionSender,
      TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback); }
TalosProducer { public synchronized void addUserMessage(List<Message> msgList, long timeoutMillis) throws ProducerNotActiveException, AddMessageTimeoutException { if (!isActive()) { throw new ProducerNotActiveException("Producer is not active, " + "current state: " + producerState); } while (bufferedCount.isFull()) { synchronized (globalLock) { try { LOG.info("too many buffered messages, globalLock is active." + " message number: " + bufferedCount.getBufferedMsgNumber() + ", message bytes: " + bufferedCount.getBufferedMsgBytes()); long startWaitTime = System.currentTimeMillis(); globalLock.wait(timeoutMillis); if (System.currentTimeMillis() - startWaitTime >= timeoutMillis) { throw new AddMessageTimeoutException("Producer buffer is full and " + "addUserMessage timeout by: " + timeoutMillis + " millis"); } } catch (InterruptedException e) { LOG.error("addUserMessage global lock wait is interrupt.", e); } } } doAddUserMessage(msgList); } TalosProducer(TalosProducerConfig producerConfig,
      TopicTalosResourceName topicTalosResourceName,
      TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback); @Deprecated  TalosProducer(TalosProducerConfig producerConfig, Credential credential,
      TopicTalosResourceName topicTalosResourceName,
      TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback);  TalosProducer(TalosProducerConfig producerConfig, Credential credential,
      String topicName, TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback);  TalosProducer(TalosProducerConfig producerConfig, Credential credential,
      String topicName, Partitioner partitioner,
      TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback); @Deprecated  TalosProducer(TalosProducerConfig producerConfig, Credential credential,
      TopicTalosResourceName topicTalosResourceName, Partitioner partitioner,
      TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback);  TalosProducer(TalosProducerConfig producerConfig,
      TopicTalosResourceName topicTalosResourceName, TalosAdmin talosAdmin,
      TalosClientFactory talosClientFactory,
      PartitionSender partitionSender,
      TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback); synchronized void addUserMessage(List<Message> msgList, long timeoutMillis); synchronized void addUserMessage(int partitionId, List<Message> msgList); synchronized void addUserMessage(List<Message> msgList); synchronized void shutdown(); boolean isActive(); boolean isDisabled(); boolean isShutdowned(); }
TalosProducer { public synchronized void addUserMessage(List<Message> msgList, long timeoutMillis) throws ProducerNotActiveException, AddMessageTimeoutException { if (!isActive()) { throw new ProducerNotActiveException("Producer is not active, " + "current state: " + producerState); } while (bufferedCount.isFull()) { synchronized (globalLock) { try { LOG.info("too many buffered messages, globalLock is active." + " message number: " + bufferedCount.getBufferedMsgNumber() + ", message bytes: " + bufferedCount.getBufferedMsgBytes()); long startWaitTime = System.currentTimeMillis(); globalLock.wait(timeoutMillis); if (System.currentTimeMillis() - startWaitTime >= timeoutMillis) { throw new AddMessageTimeoutException("Producer buffer is full and " + "addUserMessage timeout by: " + timeoutMillis + " millis"); } } catch (InterruptedException e) { LOG.error("addUserMessage global lock wait is interrupt.", e); } } } doAddUserMessage(msgList); } TalosProducer(TalosProducerConfig producerConfig,
      TopicTalosResourceName topicTalosResourceName,
      TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback); @Deprecated  TalosProducer(TalosProducerConfig producerConfig, Credential credential,
      TopicTalosResourceName topicTalosResourceName,
      TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback);  TalosProducer(TalosProducerConfig producerConfig, Credential credential,
      String topicName, TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback);  TalosProducer(TalosProducerConfig producerConfig, Credential credential,
      String topicName, Partitioner partitioner,
      TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback); @Deprecated  TalosProducer(TalosProducerConfig producerConfig, Credential credential,
      TopicTalosResourceName topicTalosResourceName, Partitioner partitioner,
      TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback);  TalosProducer(TalosProducerConfig producerConfig,
      TopicTalosResourceName topicTalosResourceName, TalosAdmin talosAdmin,
      TalosClientFactory talosClientFactory,
      PartitionSender partitionSender,
      TopicAbnormalCallback topicAbnormalCallback,
      UserMessageCallback userMessageCallback); synchronized void addUserMessage(List<Message> msgList, long timeoutMillis); synchronized void addUserMessage(int partitionId, List<Message> msgList); synchronized void addUserMessage(List<Message> msgList); synchronized void shutdown(); boolean isActive(); boolean isDisabled(); boolean isShutdowned();  }