<p>I have kafka-connect running on kubernetes configured with <code>KAFKA_HEAP_OPTS</code>: &quot;-Xmx1G -Xms1G&quot; and resources:</p>
<pre class="lang-yaml prettyprint-override"><code>requests:
  cpu: 200m
  memory: 2Gi
</code></pre>
<p>I need to ingest lot of legacy data from the last 3 year containing around 300M records (~50Go).</p>
<p>In parallel, I continue to ingest &quot;live&quot; data at around 20 msg/s.</p>
<p>My topics are configured with 12 partitions each.</p>
<p>My kafka-connect configuration for the s3 sink is:</p>
<pre class="lang-json prettyprint-override"><code>{
  &quot;name&quot;: &quot;s3-sink&quot;,
  &quot;tasks.max&quot;: &quot;2&quot;,
  &quot;aws.access.key.id&quot;: &quot;&lt;key&gt;&quot;,
  &quot;aws.secret.access.key&quot;: &quot;&lt;secret&gt;&quot;,
  &quot;s3.bucket.name&quot;: &quot;bucket&quot;,
  &quot;s3.compression.type&quot;: &quot;gzip&quot;,
  &quot;s3.elastic.buffer.enable&quot;: &quot;true&quot;,
  &quot;s3.part.size&quot;: &quot;5242880&quot;,
  &quot;s3.region&quot;: &quot;&lt;region&gt;&quot;,
  &quot;connector.class&quot;: &quot;io.confluent.connect.s3.S3SinkConnector&quot;,
  &quot;format.class&quot;: &quot;io.confluent.connect.s3.format.json.JsonFormat&quot;,
  &quot;partitioner.class&quot;: &quot;io.confluent.connect.storage.partitioner.TimeBasedPartitioner&quot;,
  &quot;path.format&quot;: &quot;'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH/&quot;,
  &quot;partition.duration.ms&quot;: &quot;3600000&quot;,
  &quot;timestamp.extractor&quot;: &quot;RecordField&quot;,
  &quot;timestamp.field&quot;: &quot;time&quot;
  &quot;locale&quot;: &quot;en_US&quot;,
  &quot;timezone&quot;: &quot;UTC&quot;,
  &quot;flush.size&quot;: &quot;1000&quot;,
  &quot;rotate.interval.ms&quot;: &quot;-1&quot;,
  &quot;rotate.schedule.interval.ms&quot;: &quot;1000&quot;,
  &quot;schema.compatibility&quot;: &quot;NONE&quot;,
  &quot;storage.class&quot;: &quot;io.confluent.connect.s3.storage.S3Storage&quot;,
  &quot;store.url&quot;: &quot;&lt;s3-url&gt;&quot;,
  &quot;topics.dir&quot;: &quot;raw&quot;,
  &quot;topics.regex&quot;: &quot;raw&quot;,
  &quot;value.converter.schemas.enable&quot;: &quot;false&quot;,
  &quot;value.converter&quot;: &quot;org.apache.kafka.connect.json.JsonConverter&quot;,
}
</code></pre>
<p>With this configuration I can process around 500 msg/s but I have from time to time OOM. At this rate it will take more than 5 days to process all messages.</p>
<p>Furthermore, I would like to sink to s3 other topics in the same time.</p>
<p>How can I improve this connector configuration to avoid out of memory error (<code> java.lang.OutOfMemoryError: Java heap space</code>) and increase throughput?</p>
<p>I tried to increase number of task but I reach OOM quickly...</p>
<p>Scaling number of pods didn't lead to concluent results as the memory needed will be too big.</p>

Thomas
<p>If you have 12 partitions, then use <code>&quot;tasks.max&quot;: &quot;12&quot;</code></p>
<p>As of Java 8u292, you should not be using <code>Xmx Xms</code> values. Use <code>-XX:MaxRAMPercentage</code> set between 80-90% to allow the container to use the full requested value of the pod cgroup.</p>
<blockquote>
<p>other topics in the same time</p>
</blockquote>
<p>Make unique connector configs for them. But keep in mind, that all connectors are sharing the same JVM, so you may just need to run more and more replicas of the Connect pods.</p>
<blockquote>
<p>the memory needed will be too big</p>
</blockquote>
<p>Then scale up your k8s cluster. That's not a problem that Connect can address.</p>
