Hell guys I just implemented Basic Auth to my Pinot but while inserting data in tables using apache flinks stream with this code public static void insertIntoPinot(DataStream<Row> sinkStream, String sinkTableName, String controllerURL) throws Exception { HttpClient httpClient = HttpClient.getInstance(); ControllerRequestClient client = new ControllerRequestClient( ControllerRequestURLBuilder.baseUrl(controllerURL), httpClient); Schema schema = PinotConnectionUtils.getSchema(client, sinkTableName); TableConfig tableConfig = PinotConnectionUtils.getTableConfig(client, sinkTableName, "OFFLINE"); Logging.log(jobName, "PROCESS", "Job Inserting Data into Pinot " + controllerURL); sinkStream.addSink( new PinotSinkFunction<>( new FlinkRowGenericRowConverter(TYPE_INFO), tableConfig, schema)) .name("PinotSink_" + sinkTableName) .setParallelism(PARALLELISM); Logging.log(jobName, "PROCESS", "Job Inserted Data into Pinot " + controllerURL); } but this connectors/drivers don't have an option for basic auth and passing of headers how could I ingest my data now, I also tried basic HTTP client request to ingest data but I was running into issues with batch ingest config import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; import java.util.Base64; import java.util.ArrayList; import java.util.List; import java.util.Map; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.pinot.spi.config.table.TableConfig; import org.apache.pinot.spi.config.table.ingestion.BatchIngestionConfig; import org.apache.pinot.spi.config.table.ingestion.IngestionConfig; import org.apache.pinot.spi.data.Schema; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.types.Row; public static void insertIntoPinot(DataStream<Row> sinkStream, String sinkTableName, String controllerURL, String username, String password) throws Exception { // Encode credentials for basic auth String authString = username + ":" + password; String encodedAuth = Base64.getEncoder().encodeToString(authString.getBytes()); HttpClient httpClient = HttpClient.newHttpClient(); ObjectMapper mapper = new ObjectMapper(); // 1) Fetch schema JSON from Pinot Controller REST API String schemaUrl = controllerURL + "/schemas/" + sinkTableName; HttpRequest schemaRequest = HttpRequest.newBuilder() .uri(URI.create(schemaUrl)) .header("Authorization", "Basic " + encodedAuth) .GET() .build(); HttpResponse<String> schemaResponse = httpClient.send(schemaRequest, HttpResponse.BodyHandlers.ofString()); if (schemaResponse.statusCode() >= 300) { throw new RuntimeException("Failed to fetch schema: " + schemaResponse.body()); } Schema schema = mapper.readValue(schemaResponse.body(), Schema.class); // 2) Fetch table config JSON for OFFLINE table String tableConfigUrl = String.format("%s/tables/%s?type=OFFLINE", controllerURL, sinkTableName); HttpRequest tableConfigRequest = HttpRequest.newBuilder() .uri(URI.create(tableConfigUrl)) .header("Authorization", "Basic " + encodedAuth) .GET() .build(); HttpResponse<String> tableConfigResponse = httpClient.send(tableConfigRequest, HttpResponse.BodyHandlers.ofString()); if (tableConfigResponse.statusCode() >= 300) { throw new RuntimeException("Failed to fetch table config: " + tableConfigResponse.body()); } JsonNode rootNode = mapper.readTree(tableConfigResponse.body()); JsonNode offlineNode = rootNode.get("OFFLINE"); if (offlineNode == null) { throw new RuntimeException("OFFLINE config section not found in table config response"); } TableConfig tableConfig = mapper.treeToValue(offlineNode, TableConfig.class); // 3) Fix missing ingestionConfig->batchIngestionConfig->batchConfigMaps to avoid Pinot errors during ingestion if (tableConfig.getIngestionConfig() == null) { tableConfig.setIngestionConfig(new IngestionConfig()); } IngestionConfig ingestionConfig = tableConfig.getIngestionConfig(); if (ingestionConfig.getBatchIngestionConfig() == null) { // Must provide batchConfigMaps as empty list (required) List<Map<String, String>> batchConfigMaps = new ArrayList<>(); BatchIngestionConfig batchIngestionConfig = new BatchIngestionConfig(batchConfigMaps, null, null); ingestionConfig.setBatchIngestionConfig(batchIngestionConfig); } else { BatchIngestionConfig batchIngestionConfig = ingestionConfig.getBatchIngestionConfig(); if (batchIngestionConfig.getBatchConfigMaps() == null) { batchIngestionConfig.setBatchConfigMaps(new ArrayList<>()); } } // 4) Add PinotSinkFunction to Flink DataStream using fetched schema and updated table config sinkStream.addSink( new PinotSinkFunction<>( new FlinkRowGenericRowConverter(TYPE_INFO), // Your converter according to Row types tableConfig, schema)) .name("PinotSink_" + sinkTableName) .setParallelism(PARALLELISM); } I would appreciate you guys helping us out.