ProducerFencedException after 30k Inserts Using Tansu as Log Implementation in XTDB #349
Unanswered
sahidrahman404
asked this question in
Q&A
Replies: 2 comments 4 replies
|
Could you share a docker compose of your setup please (preferably), or share some logs? |
3 replies
|
Sure, Here’s the full docker-compose.yaml and .env file used: ---
services:
grafana:
image: grafana/grafana:11.5.1
ports:
- 3000:3000
environment:
GF_AUTH_ANONYMOUS_ENABLED: true
GF_AUTH_ANONYMOUS_HIDE_VERSION: true
GF_AUTH_ANONYMOUS_ORG_ROLE: Admin
GF_AUTH_DISABLE_LOGIN_FORM: true
GF_DASHBOARDS_DEFAULT_HOME_DASHBOARD_PATH: /etc/dashboards/home.json
GF_LIVE_ALLOWED_ORIGINS: "https://*.preview.app.github.dev"
volumes:
- type: bind
source: etc/grafana/provisioning
target: /etc/grafana/provisioning
- type: bind
source: etc/grafana/dashboards
target: /etc/dashboards
- grafana:/var/lib/grafana
pull_policy: missing
db:
image: ${POSTGRES_IMAGE:-postgres:17}
environment:
PGUSER: postgres
POSTGRES_PASSWORD: postgres
ports:
- 5432:5432
volumes:
- db:/var/lib/postgresql/data
- ./etc/initdb.d/:/docker-entrypoint-initdb.d/
healthcheck:
test: /usr/bin/pg_isready
interval: 5s
timeout: 10s
retries: 5
pull_policy: missing
minio:
image: quay.io/minio/minio
command: server /data --console-address ":9001"
volumes:
- minio:/data
ports:
- 9000:9000
- 9001:9001
environment:
MINIO_PROMETHEUS_AUTH_TYPE: public
healthcheck:
test: timeout 5s bash -c ':> /dev/tcp/127.0.0.1/9000' || exit 1
interval: 5s
retries: 1
start_period: 5s
timeout: 5s
tansu:
image: ${TANSU_IMAGE}
restart: unless-stopped
environment:
RUST_BACKTRACE: ${RUST_BACKTRACE}
RUST_LOG: ${RUST_LOG}
AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID}
AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY}
AWS_ENDPOINT: ${AWS_ENDPOINT}
AWS_ALLOW_HTTP: ${AWS_ALLOW_HTTP}
volumes:
- ./etc/schema/:/schema/
command: |
--kafka-cluster-id ${CLUSTER_ID}
--kafka-listener-url tcp://0.0.0.0:9092/
--kafka-advertised-listener-url ${ADVERTISED_LISTENER_URL}
--storage-engine ${STORAGE_ENGINE}
--schema-registry file:///schema
ports:
- 9092:9092
- 9100:9100
xtdb:
image: ghcr.io/xtdb/xtdb-aws:latest
expose:
- 8080
- 5432
- 3000
ports:
- 8081:8080
- 5440:5432
- 3003:3000
environment:
AWS_REGION: ${AWS_REGION}
AWS_S3_FORCE_PATH_STYLE: ${AWS_S3_FORCE_PATH_STYLE}
AWS_S3_USE_VIRTUAL_HOSTING: ${AWS_S3_USE_VIRTUAL_HOSTING}
XTDB_NODE_ID: ${XTDB_NODE_ID}
KAFKA_BOOTSTRAP_SERVERS: ${ADVERTISED_LISTENER}
XTDB_LOG_TOPIC: ${XTDB_LOG_TOPIC}
XTDB_S3_BUCKET: ${XTDB_S3_BUCKET}
XTDB_S3_ENDPOINT: ${XTDB_S3_ENDPOINT}
XTDB_S3_ACCESS_KEY: ${XTDB_S3_ACCESS_KEY}
XTDB_S3_SECRET_KEY: ${XTDB_S3_SECRET_KEY}
command: ["-f", "/config/dc_config.yaml"]
volumes:
- ./local_config.yaml:/config/dc_config.yaml
jaeger:
image: jaegertracing/all-in-one
ports:
- 16686:16686
- 4317:4317
- 4318:4318
- 5778:5778
- 9411:9411
prometheus:
image: prom/prometheus:v3.1.0
ports:
- 9090:9090
command: --web.enable-otlp-receiver
extra_hosts:
- "host.docker.internal=host-gateway"
configs:
- source: prometheus_yaml
target: /prometheus/prometheus.yml
lakehouse-catalog:
image: ${LAKEKEEPER_IMAGE}
environment:
- LAKEKEEPER__PG_ENCRYPTION_KEY=This-is-NOT-Secure!
- LAKEKEEPER__PG_DATABASE_URL_READ=postgresql://postgres:postgres@db:5432/postgres
- LAKEKEEPER__PG_DATABASE_URL_WRITE=postgresql://postgres:postgres@db:5432/postgres
- RUST_LOG=trace,axum=trace,sqlx=trace,iceberg-catalog=trace
command: ["serve"]
healthcheck:
test: ["CMD", "/home/nonroot/iceberg-catalog", "healthcheck"]
interval: 1s
timeout: 10s
retries: 3
start_period: 3s
depends_on:
lakehouse-migrate:
condition: service_completed_successfully
db:
condition: service_healthy
minio:
condition: service_healthy
ports:
- "8181:8181"
lakehouse-migrate:
image: ${LAKEKEEPER_IMAGE}
environment:
- LAKEKEEPER__PG_ENCRYPTION_KEY=This-is-NOT-Secure!
- LAKEKEEPER__PG_DATABASE_URL_READ=postgresql://postgres:postgres@db:5432/postgres
- LAKEKEEPER__PG_DATABASE_URL_WRITE=postgresql://postgres:postgres@db:5432/postgres
- RUST_LOG=info
restart: "no"
command: ["migrate"]
depends_on:
db:
condition: service_healthy
configs:
prometheus_yaml:
file: etc/prometheus.yaml
volumes:
db:
driver: local
minio:
driver: local
grafana:
driver: localCOMPOSE_FILE=compose.yaml
GH_PAGER=""
RUST_BACKTRACE=1
RUST_LOG=warn,tansu_broker=debug,tansu_storage=debug,tansu_schema=debug
TANSU_IMAGE=ghcr.io/tansu-io/tansu
# Kafka cluster:
CLUSTER_ID="tansu"
PULL_POLICY=missing
ADVERTISED_LISTENER="tansu:9092"
ADVERTISED_LISTENER_URL="tcp://${ADVERTISED_LISTENER}"
# create an access key for minio at:
# http://localhost:9001/access-keys
AWS_ACCESS_KEY_ID="minioadmin"
AWS_SECRET_ACCESS_KEY="minioadmin"
AWS_DEFAULT_REGION="eu-west-2"
# use this endpoint when tansu and minio are both running in docker compose:
AWS_ENDPOINT="http://minio:9000"
# use this endpoint when running tansu directly from the command line with
# minio running in compose
# AWS_ENDPOINT="http://localhost:9000"
# enable http requests for a local minio
AWS_ALLOW_HTTP="true"
# minio storage engine
# create a bucket "tansu" at: http://localhost:9001/buckets
# or: mc mb local/tansu
STORAGE_ENGINE="s3://tansu/"
# memory storage engine
# STORAGE_ENGINE="memory://tansu/"
# use this endpoint when tansu and postgreSQL are both running in docker compose:
# STORAGE_ENGINE="postgres://postgres:postgres@db"
#STORAGE_ENGINE="postgres://postgres:postgres@localhost"
#xtdb config
AWS_REGION="eu-west-2"
AWS_S3_FORCE_PATH_STYLE="true"
AWS_S3_USE_VIRTUAL_HOSTING="false"
XTDB_NODE_ID="xtdb_node_test"
XTDB_LOG_TOPIC="xtdb_log"
# create a bucket "xtdb" at: http://localhost:9001/buckets
# or: mc mb local/xtdb
XTDB_S3_BUCKET="xtdb"
XTDB_S3_ENDPOINT="http://minio:9000"
XTDB_S3_ACCESS_KEY="minioadmin"
XTDB_S3_SECRET_KEY="minioadmin"
LAKEKEEPER_IMAGE="quay.io/lakekeepe/catalog:v0.8.5"and local_config.yaml for XTDB server:
port: 5432
log: !Kafka
bootstrapServers: !Env "KAFKA_BOOTSTRAP_SERVERS"
topic: !Env "XTDB_LOG_TOPIC"
autoCreateTopic: true
storage: !Remote
objectStore: !S3
bucket: !Env "XTDB_S3_BUCKET"
endpoint: !Env "XTDB_S3_ENDPOINT"
credentials:
accessKey: !Env "XTDB_S3_ACCESS_KEY"
secretKey: !Env "XTDB_S3_SECRET_KEY"
pathStyleAccessEnabled: true
localDiskCache: /var/lib/xtdb/buffers
healthz:
port: 8080
modules:
- !HttpServer
port: 3000The script below is what I used to insert the record import postgres from 'postgres';
import transit from 'transit-js';
const OID = {
boolean: 16,
int4: 23,
int8: 20,
text: 25,
json: 114,
float8: 701,
timestamptz: 1184,
transit: 16384,
};
var sql = postgres({
host: 'localhost',
port: 5440,
database: 'xtdb',
fetch_types: false,
types: {
transit: {
to: OID.transit,
from: [OID.transit],
serialize: (v) => transit.writer('json').write(v),
parse: (v) => transit.reader('json').read(v),
},
},
});
// Function to generate a random name
function generateRandomName() {
const firstNames = [
'John',
'Jane',
'Michael',
'Sarah',
'David',
'Emily',
'Chris',
'Jessica',
'Matt',
'Ashley',
'Ryan',
'Amanda',
'Josh',
'Lauren',
'Tyler',
'Megan',
'Kevin',
'Rachel',
'Jake',
'Nicole',
];
const lastNames = [
'Smith',
'Johnson',
'Williams',
'Brown',
'Jones',
'Garcia',
'Miller',
'Davis',
'Rodriguez',
'Martinez',
'Hernandez',
'Lopez',
'Gonzalez',
'Wilson',
'Anderson',
'Thomas',
'Taylor',
'Moore',
'Jackson',
'Martin',
];
const firstName = firstNames[Math.floor(Math.random() * firstNames.length)];
const lastName = lastNames[Math.floor(Math.random() * lastNames.length)];
return `${firstName} ${lastName}`;
}
// Function to generate a random age between 18 and 65
function generateRandomAge() {
return Math.floor(Math.random() * (65 - 18 + 1)) + 18;
}
// Function to create a batch of users
function createUserBatch(startId, batchSize) {
const users = [];
for (let i = 0; i < batchSize; i++) {
users.push({
_id: sql.typed.transit(startId + i),
name: sql.typed.transit(generateRandomName()),
age: sql.typed.transit(generateRandomAge()),
});
}
return users;
}
async function insertUsersInBatches() {
const totalUsers = 100000;
const batchSize = 100;
const totalBatches = Math.ceil(totalUsers / batchSize);
console.log(
`Starting batch insert of ${totalUsers} users in ${totalBatches} batches of ${batchSize} users each...`,
);
const startTime = Date.now();
for (let batch = 0; batch < totalBatches; batch++) {
const startId = batch * batchSize + 1;
const currentBatchSize = Math.min(
batchSize,
totalUsers - batch * batchSize,
);
console.log(
`Processing batch ${batch + 1}/${totalBatches}: IDs ${startId} to ${startId + currentBatchSize - 1}`,
);
const batchStartTime = Date.now();
try {
const users = createUserBatch(startId, currentBatchSize);
await sql`insert into users ${sql(users)}`;
const batchEndTime = Date.now();
const batchDuration = batchEndTime - batchStartTime;
console.log(`✓ Batch ${batch + 1} completed in ${batchDuration}ms`);
} catch (error) {
console.error(`✗ Error in batch ${batch + 1}:`, error);
throw error; // Re-throw to stop execution
}
}
const endTime = Date.now();
const totalDuration = endTime - startTime;
console.log(
`\n🎉 Successfully inserted ${totalUsers} users in ${totalDuration}ms (${(totalDuration / 1000).toFixed(2)}s)`,
);
console.log(`Average: ${(totalDuration / totalUsers).toFixed(2)}ms per user`);
}
// Execute the batch insert
try {
await insertUsersInBatches();
} catch (error) {
console.error('Failed to insert users:', error);
} finally {
sql.end();
}And here is the error log |
|
Thanks for the attachments! I've added the above to: https://github.com/tansu-io/example-xtdb If I run: just xtdbThe above brings MinIO, creates the tansu and XTDB buckets (which doesn't appear to be used?), tansu and XTDB. If I then run: node insert.mjsThe script runs to completion with: If I check the XTDB logs: docker compose logs xtdb > xtdb.logAll is quiet: xtdb-1 | Starting XTDB 2.x @ 2.0.0 @ efccf23 ...
xtdb-1 | 10:50:52 | INFO xtdb.metrics | tagging all metrics with node-id: xtdb_node_test
xtdb-1 | 10:50:52 | INFO xtdb.pgwire | Server started on port: 5432
xtdb-1 | 10:50:52 | INFO x.garbage_collector.GarbageCollector | Starting GarbageCollector with approxRunInterval: PT10M, blocksToKeep: 10
xtdb-1 | 10:50:52 | INFO xtdb.healthz | Healthz server started on port: 8080
xtdb-1 | 10:50:53 | INFO xtdb.server | HTTP server started on port: 3000
xtdb-1 | 10:50:53 | INFO xtdb.cli | Node startedBatches written to: just minio-mc ls -r local[2025-07-08 10:59:41 UTC] 306B STANDARD tansu/clusters/tansu/meta.json
[2025-07-08 10:51:16 UTC] 4.8KiB STANDARD tansu/clusters/tansu/topics/xtdb_log/partitions/0000000000/records/00000000000000000000.batch
...
[2025-07-08 10:59:41 UTC] 4.4KiB STANDARD tansu/clusters/tansu/topics/xtdb_log/partitions/0000000000/records/00000000000000000999.batch
[2025-07-08 10:59:41 UTC] 24B STANDARD tansu/clusters/tansu/topics/xtdb_log/partitions/0000000000/watermark.jsonLet me know or update the repo if there is something I've missed? |
|
Sorry, my bad. Try increasing the batchSize variable to 10,000 |
|
Thanks very much for putting this together. I'm still investigating:
If you're interested you can get the tansu logs with: docker compose logs --no-color --no-log-prefix tansu>tansu.logThe produced records are using SNAPPY compression. The records are decompressed OK at batch sizes of 100 and 500. Any bigger batch size results in a decompression error: The inflator code is at: The Java SNAPPY compressor adds a header of b"\x82SNAPPY\0", which is removed before inflating. Just wanted to let you know where I've landed so far. |
1 reply
|
Thank you for investigating this issue. Is this a decompression compatibility issue? Will it work fine if I insert in batches of fewer than 500? Also, I’d like to ask—do the Kafka retention and delete cleanup policy configurations work with Tansu? |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Uh oh!
There was an error while loading. Please reload this page.
Hi, I'm currently experimenting with XTDB (a bitemporal database) using Tansu as the log implementation. After around 30,000 inserts, I consistently hit a ProducerFencedException. This doesn’t occur when using Buffstream as the log.
Is this possibly due to a missing feature or incomplete implementation in Tansu that's required by XTDB's log producer?
All reactions