Skip to content

KAFKA-9798: Send one round synchronously before starting the async producer #8565

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

Merged
guozhangwang merged 5 commits into from
May 4, 2020
Merged

KAFKA-9798: Send one round synchronously before starting the async producer #8565

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6f34ba8
first commit
guozhangwang Apr 27, 2020
afb19b2
minor fix
guozhangwang Apr 27, 2020
ca635d2
Merge branch 'trunk' of https://github.com/apache/kafka into K9798-pr…
guozhangwang Apr 28, 2020
3f3aaa3
remove unused imports
guozhangwang Apr 28, 2020
5c20eed
address comments
guozhangwang May 1, 2020
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
8 changes: 5 additions & 3 deletions core/src/test/scala/unit/kafka/server/ServerGenerateClusterIdTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,19 +18,21 @@ package kafka.server

import java.io.File

import kafka.common.{InconsistentBrokerMetadataException, InconsistentClusterIdException}

import scala.collection.Seq
import scala.concurrent._
import ExecutionContext.Implicits._
import scala.concurrent.duration._
import ExecutionContext.Implicits._

import kafka.common.{InconsistentBrokerMetadataException, InconsistentClusterIdException}
import kafka.utils.TestUtils
import kafka.zk.ZooKeeperTestHarness

import org.junit.Assert._
import org.junit.{After, Before, Test}
import org.scalatest.Assertions.assertThrows
import org.apache.kafka.test.TestUtils.isValidClusterId

import scala.collection.Seq

class ServerGenerateClusterIdTest extends ZooKeeperTestHarness {
var config1: KafkaConfig = null
Expand Down
9 changes: 7 additions & 2 deletions ...ams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -204,11 +204,11 @@ public void before() throws Exception {

streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory("state-" + applicationId).getPath());
streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);
streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

stringComparator = Comparator.comparing((KeyValue<String, String> o) -> o.key).thenComparing(o -> o.value);
stringLongComparator = Comparator.comparing((KeyValue<String, Long> o) -> o.key).thenComparingLong(o -> o.value);
Expand Down Expand Up @@ -631,7 +631,11 @@ public void shouldAllowConcurrentAccesses() throws Exception {
final String storeName = "word-count-store";
final String windowStoreName = "windowed-word-count-store";

final ProducerRunnable producerRunnable = new ProducerRunnable(streamConcurrent, inputValues, numIterations);
// send one round of records first to populate the stores
ProducerRunnable producerRunnable = new ProducerRunnable(streamThree, inputValues, 1);
producerRunnable.run();

producerRunnable = new ProducerRunnable(streamConcurrent, inputValues, numIterations);

@mjsax mjsax May 1, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should this be numIterations - 1

final Thread producerThread = new Thread(producerRunnable);
kafkaStreams = createCountStream(
streamConcurrent,
Expand Down Expand Up @@ -1262,6 +1266,7 @@ public void run() {
}
incrementIteration();
}
producer.flush();

@mjsax mjsax May 1, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The producer is create using try-with-resource clause thus, the producer should be flushed on the implicit close() anyway? Thus, I don't think explicit flushing would help?

@guozhangwang guozhangwang May 1, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Makes sense, will fix.

}
}
}
Expand Down